diff --git a/crates/notify/src/config_manager.rs b/crates/notify/src/config_manager.rs index 7cd83ffe3..fab815865 100644 --- a/crates/notify/src/config_manager.rs +++ b/crates/notify/src/config_manager.rs @@ -228,6 +228,18 @@ impl NotifyConfigManager { let ttype = target_type.to_lowercase(); let tname = target_id.id.to_lowercase(); + // Guard against orphaning bucket notification rules (backlog#979). Removing a + // target while a bucket rule still references it would leave a dangling + // binding whose events can never be delivered. This mirrors the symmetric + // guard already applied in `remove_target_config`: refuse the removal while + // the target is still bound so the caller unbinds the bucket rules first. + let bound_target_id = runtime_target_id_for_subsystem(&ttype, &tname); + if self.rule_engine.is_target_bound_to_any_bucket(&bound_target_id).await { + return Err(NotificationError::Configuration(format!( + "Target is still bound to bucket rules and deletion is prohibited: type={ttype} name={tname}" + ))); + } + self.update_config_and_reload(|config| { let mut changed = false; if let Some(targets_of_type) = config.0.get_mut(&ttype) { @@ -346,6 +358,17 @@ impl NotifyConfigManager { self.update_config(new_config.clone()).await; + // Stop the currently running replay workers *before* activating the new ones + // (backlog#970). Each replay worker drains a per-target persisted store; if the + // new workers start while the old ones are still running against the same + // stores, both drain the same queues and re-deliver events. `replace_targets` + // below also stops workers, but only after `activate_targets_with_replay` has + // already spawned the new ones — so without this explicit stop-before-start + // there is a window where old and new workers overlap. (The full "signal + + // join" shutdown lives in the targets crate and is tracked under the same + // issue; this reorders the notify-side lifecycle.) + self.runtime_facade.stop_replay_workers().await; + let targets: Vec + Send + Sync>> = self .registry .create_targets_from_config(&new_config) @@ -427,6 +450,8 @@ impl NotifyConfigManager { #[cfg(test)] mod tests { use super::{NotifyConfigManager, NotifyConfigStoreError, runtime_target_id_for_subsystem, serialized_read_modify_write}; + use crate::NotificationError; + use crate::rules::RulesMap; use crate::{ integration::NotificationMetrics, notifier::EventNotifier, registry::TargetRegistry, rule_engine::NotifyRuleEngine, runtime_facade::NotifyRuntimeFacade, @@ -436,7 +461,9 @@ mod tests { NOTIFY_PULSAR_SUB_SYS, NOTIFY_REDIS_SUB_SYS, NOTIFY_WEBHOOK_SUB_SYS, }; use rustfs_config::server_config::{Config, KVS}; + use rustfs_s3_types::EventName; use rustfs_targets::ReplayWorkerManager; + use rustfs_targets::arn::TargetID; use std::sync::Arc; use tokio::sync::{RwLock, Semaphore}; @@ -457,6 +484,33 @@ mod tests { NotifyConfigManager::new(config, registry, rule_engine, runtime_facade) } + // Regression test for backlog#979 (b): `remove_target` must refuse to remove a + // target that is still referenced by a bucket notification rule, otherwise the + // rule is left orphaned (pointing at a target that no longer exists). This + // mirrors the guard `remove_target_config` already enforces. The guard runs + // before the persisted config read-modify-write, so it returns the refusal + // without needing a live object store. + #[tokio::test] + async fn remove_target_refuses_when_bound_to_bucket_rules() { + let manager = build_manager(); + + // Bind the runtime target id (webhook/primary) that `remove_target` derives + // from (`NOTIFY_WEBHOOK_SUB_SYS`, id="primary") to a bucket rule. + let bound_id = TargetID::new("primary".to_string(), "webhook".to_string()); + let mut rules_map = RulesMap::new(); + rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), bound_id.clone()); + manager.rule_engine.set_bucket_rules("bucket", rules_map).await; + + let err = manager + .remove_target(&bound_id, NOTIFY_WEBHOOK_SUB_SYS) + .await + .expect_err("removing a bound target must be refused to avoid orphaning bucket rules"); + assert!( + matches!(err, NotificationError::Configuration(_)), + "expected a Configuration refusal, got {err:?}" + ); + } + #[tokio::test] async fn config_manager_init_accepts_empty_target_set() { let manager = build_manager(); diff --git a/crates/notify/src/event.rs b/crates/notify/src/event.rs index 7d9ef17c2..455c5f7c8 100644 --- a/crates/notify/src/event.rs +++ b/crates/notify/src/event.rs @@ -249,7 +249,19 @@ impl Event { let key_name = form_urlencoded::byte_serialize(args.object.name.as_bytes()).collect::(); let principal_id = args.req_params.get("principalId").unwrap_or(&String::new()).to_string(); - let version_id = args.object.version_id.clone().or_else(|| Some(args.version_id.clone())); + // An unversioned object must omit `versionId` entirely rather than emit it as + // an empty string. Prefer the object's own version id, fall back to the + // request-scoped one, and treat an empty value from either source as "no + // version" so serialization skips the field (`skip_serializing_if` on + // `Object::version_id`). This matches S3 and the repo's tier convention that a + // `None`/`""` version means "unversioned" (backlog#984). + let version_id = args + .object + .version_id + .clone() + .filter(|v| !v.is_empty()) + .or_else(|| Some(args.version_id.clone())) + .filter(|v| !v.is_empty()); let mut s3_metadata = Metadata { schema_version: "1.0".to_string(), @@ -619,6 +631,68 @@ mod tests { assert_eq!(user_metadata.len(), 2); } + #[test] + fn unversioned_object_omits_version_id() { + // Neither the object nor the request carries a version id: the field must be + // `None` so it is omitted from the serialized event, not `Some("")` (backlog#984). + let args = EventArgsBuilder::new( + EventName::ObjectCreatedPut, + "bucket", + NotifyObjectInfo { + bucket: "bucket".to_string(), + name: "key".to_string(), + version_id: None, + ..Default::default() + }, + ) + .version_id(String::new()) + .build(); + let event = Event::new(args); + assert_eq!(event.s3.object.version_id, None); + + let json = serde_json::to_value(&event).expect("event should serialize"); + assert!( + json.pointer("/s3/object/versionId").is_none(), + "unversioned object must not serialize a versionId field" + ); + } + + #[test] + fn empty_object_version_falls_back_then_omits() { + // Empty object version must not shadow a real request-scoped version. + let args = EventArgsBuilder::new( + EventName::ObjectCreatedPut, + "bucket", + NotifyObjectInfo { + bucket: "bucket".to_string(), + name: "key".to_string(), + version_id: Some(String::new()), + ..Default::default() + }, + ) + .version_id("v-42".to_string()) + .build(); + let event = Event::new(args); + assert_eq!(event.s3.object.version_id.as_deref(), Some("v-42")); + } + + #[test] + fn present_object_version_is_preserved() { + let args = EventArgsBuilder::new( + EventName::ObjectCreatedPut, + "bucket", + NotifyObjectInfo { + bucket: "bucket".to_string(), + name: "key".to_string(), + version_id: Some("v-1".to_string()), + ..Default::default() + }, + ) + .build(); + let event = Event::new(args); + assert_eq!(event.s3.object.version_id.as_deref(), Some("v-1")); + } + #[test] fn event_time_serializes_with_millisecond_precision() { let mut event = Event::new_test_event("bucket", "key", EventName::ObjectCreatedPut); diff --git a/crates/notify/src/global.rs b/crates/notify/src/global.rs index 894c51e30..6eab1e73f 100644 --- a/crates/notify/src/global.rs +++ b/crates/notify/src/global.rs @@ -37,7 +37,14 @@ pub async fn initialize(config: Config) -> Result<(), NotificationError> { match NOTIFICATION_SYSTEM.set(Arc::new(system)) { Ok(_) => Ok(()), - Err(_) => Err(NotificationError::Lifecycle(LifecycleError::AlreadyInitialized)), + Err(losing_system) => { + // Another initializer won the race. `init()` above already started this + // system's targets and replay workers, so simply dropping it would leak + // those background tasks. Shut the losing instance down cleanly before + // reporting the conflict (backlog#984). + losing_system.shutdown().await; + Err(NotificationError::Lifecycle(LifecycleError::AlreadyInitialized)) + } } } diff --git a/crates/notify/src/notifier.rs b/crates/notify/src/notifier.rs index 601a50dad..2fce24308 100644 --- a/crates/notify/src/notifier.rs +++ b/crates/notify/src/notifier.rs @@ -31,6 +31,36 @@ const EVENT_NOTIFY_RUNTIME_LIFECYCLE: &str = "notify_runtime_lifecycle"; pub type SharedNotifyTargetList = Arc>; +/// Resolves the effective send concurrency (semaphore permit count). +/// +/// A value of `0` would build a zero-permit semaphore, so `acquire` never +/// completes and every dispatch deadlocks. A misconfigured +/// `RUSTFS_NOTIFY_SEND_CONCURRENCY=0` therefore coerces back to the default +/// instead of silently wedging notifications (backlog#984). +fn resolve_send_concurrency() -> usize { + let configured = rustfs_utils::get_env_usize(ENV_NOTIFY_SEND_CONCURRENCY, DEFAULT_NOTIFY_SEND_CONCURRENCY); + coerce_send_concurrency(configured) +} + +/// Coerces a configured send concurrency into a valid semaphore permit count. +/// `0` maps back to the default; any positive value is passed through. Kept as a +/// pure function so the coercion is unit-testable without mutating process env. +fn coerce_send_concurrency(configured: usize) -> usize { + if configured == 0 { + warn!( + event = EVENT_NOTIFY_RUNTIME_LIFECYCLE, + component = LOG_COMPONENT_NOTIFY, + subsystem = LOG_SUBSYSTEM_DISPATCH, + configured, + default = DEFAULT_NOTIFY_SEND_CONCURRENCY, + "invalid RUSTFS_NOTIFY_SEND_CONCURRENCY=0; falling back to default to avoid a zero-permit deadlock" + ); + DEFAULT_NOTIFY_SEND_CONCURRENCY + } else { + configured + } +} + /// Manages event notification to targets based on rules pub struct EventNotifier { metrics: Arc, @@ -51,7 +81,7 @@ impl EventNotifier { /// # Returns /// Returns a new instance of EventNotifier. pub fn new(metrics: Arc, rule_engine: NotifyRuleEngine) -> Self { - let max_inflight = rustfs_utils::get_env_usize(ENV_NOTIFY_SEND_CONCURRENCY, DEFAULT_NOTIFY_SEND_CONCURRENCY); + let max_inflight = resolve_send_concurrency(); EventNotifier { metrics, rule_engine, @@ -273,7 +303,10 @@ impl EventNotifier { #[instrument(skip(self, targets_to_init))] pub async fn init_bucket_targets_shared(&self, targets_to_init: Vec>) -> Result<(), NotificationError> { let mut target_list_guard = self.target_list.write().await; - target_list_guard.clear(); + // Close the currently registered targets before replacing them. A bare + // `clear()` drops the old targets without invoking `close()`, leaking their + // connections/streams; `clear_targets_only` closes each one first (backlog#984). + target_list_guard.clear_targets_only().await; for target in targets_to_init { debug!( @@ -727,4 +760,112 @@ mod tests { assert_ne!(metrics.processing_count(), usize::MAX); assert_eq!(metrics.processed_count(), 1); } + + /// Regression test for backlog#984: a `RUSTFS_NOTIFY_SEND_CONCURRENCY=0` + /// misconfiguration must not build a zero-permit semaphore (which would + /// deadlock every dispatch); it must fall back to the default. Positive values + /// pass through unchanged. + #[test] + fn zero_send_concurrency_falls_back_to_default() { + assert_eq!( + coerce_send_concurrency(0), + DEFAULT_NOTIFY_SEND_CONCURRENCY, + "zero concurrency must fall back to the default, never a zero-permit semaphore" + ); + assert!(coerce_send_concurrency(0) > 0, "resolved concurrency must be strictly positive"); + assert_eq!(coerce_send_concurrency(1), 1); + assert_eq!(coerce_send_concurrency(128), 128); + } + + /// Regression test for backlog#984: replacing the runtime target set via + /// `init_bucket_targets_shared` must `close()` the previously registered + /// targets instead of dropping them silently (connection/stream leak). + #[tokio::test] + async fn init_bucket_targets_shared_closes_replaced_targets() { + let rule_engine = NotifyRuleEngine::new(); + let notifier = EventNotifier::new(Arc::new(NotificationMetrics::new()), rule_engine); + + let old_target = ClosableTestTarget::new("old", "webhook"); + notifier + .init_bucket_targets_shared(vec![Arc::new(old_target.clone()) as SharedTarget]) + .await + .expect("initial install should succeed"); + assert_eq!(old_target.close_calls.load(Ordering::SeqCst), 0, "target must not close on first install"); + + let new_target = ClosableTestTarget::new("new", "webhook"); + notifier + .init_bucket_targets_shared(vec![Arc::new(new_target.clone()) as SharedTarget]) + .await + .expect("replacement install should succeed"); + + assert_eq!( + old_target.close_calls.load(Ordering::SeqCst), + 1, + "the replaced target must be closed exactly once" + ); + assert_eq!( + new_target.close_calls.load(Ordering::SeqCst), + 0, + "the freshly installed target must stay open" + ); + } + + /// A target that records `close()` invocations, for lifecycle assertions. + #[derive(Clone)] + struct ClosableTestTarget { + id: TargetID, + close_calls: Arc, + } + + impl ClosableTestTarget { + fn new(id: &str, name: &str) -> Self { + Self { + id: TargetID::new(id.to_string(), name.to_string()), + close_calls: Arc::new(AtomicUsize::new(0)), + } + } + } + + #[async_trait] + impl Target for ClosableTestTarget + where + E: rustfs_targets::PluginEvent, + { + fn id(&self) -> TargetID { + self.id.clone() + } + + async fn is_active(&self) -> Result { + Ok(true) + } + + async fn save(&self, _event: Arc>) -> Result<(), TargetError> { + Ok(()) + } + + async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { + Ok(()) + } + + async fn close(&self) -> Result<(), TargetError> { + self.close_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + + fn store(&self) -> Option<&(dyn Store + Send + Sync)> { + None + } + + fn clone_dyn(&self) -> Box + Send + Sync> { + Box::new(self.clone()) + } + + async fn init(&self) -> Result<(), TargetError> { + Ok(()) + } + + fn is_enabled(&self) -> bool { + true + } + } } diff --git a/crates/notify/src/pipeline.rs b/crates/notify/src/pipeline.rs index 3d42ea8b5..b2e8d29f2 100644 --- a/crates/notify/src/pipeline.rs +++ b/crates/notify/src/pipeline.rs @@ -103,8 +103,18 @@ impl NotifyPipeline { } pub async fn send_event(&self, event: Arc) { - self.live_event_history.write().await.record(event.clone()); - let _ = self.live_event_sender.send(event.clone()); + // Assign the history sequence number and publish to live subscribers under + // the *same* critical section. `record` allocates a monotonically increasing + // sequence and `broadcast::Sender::send` is synchronous, so holding the write + // guard across both guarantees the broadcast order matches the recorded + // sequence order. Otherwise two concurrent senders could record seq 1 then 2 + // but broadcast 2 then 1, so a cursor-based consumer would observe events out + // of order relative to the replayable history (backlog#984 / #969). + { + let mut history = self.live_event_history.write().await; + history.record(event.clone()); + let _ = self.live_event_sender.send(event.clone()); + } self.notifier.send(event).await; } } diff --git a/crates/notify/src/rules/subscriber_index.rs b/crates/notify/src/rules/subscriber_index.rs index 853aba330..88a6b57ab 100644 --- a/crates/notify/src/rules/subscriber_index.rs +++ b/crates/notify/src/rules/subscriber_index.rs @@ -93,12 +93,17 @@ impl SubscriberIndex { pub fn store_snapshot(&self, bucket: &str, new_snapshot: BucketRulesSnapshot) { let key = bucket.to_string(); - let cell = self.inner.get(&key).unwrap_or_else(|| { - // Insert a default cell (empty snapshot) - let init = Arc::new(ArcSwap::from_pointee(BucketRulesSnapshot::empty(self.empty_rules.clone()))); - self.inner.insert(key.clone(), init.clone()); - init - }); + // Atomic get-or-create of the bucket's cell. The previous `get()` then + // `insert()` was a TOCTOU: two concurrent first-writers for the same bucket + // could both observe `None`, each build a distinct `ArcSwap` cell, and both + // `insert` — the second insert overwrites the first cell, so the snapshot + // stored into the discarded cell is silently lost. `compute_if_absent` holds + // the shard write lock across the check-and-insert, so every caller shares the + // one winning cell and no snapshot is dropped (backlog#984). + let empty_rules = self.empty_rules.clone(); + let cell = self + .inner + .compute_if_absent(key, || Arc::new(ArcSwap::from_pointee(BucketRulesSnapshot::empty(empty_rules.clone())))); cell.store(Arc::new(new_snapshot)); } @@ -116,16 +121,74 @@ impl SubscriberIndex { impl Default for SubscriberIndex { fn default() -> Self { - // An available empty rule container is required; here it is implemented using minimal empty - #[derive(Debug)] - struct EmptyRules; - impl crate::rules::subscriber_snapshot::RulesContainer for EmptyRules { - type Rule = dyn crate::rules::subscriber_snapshot::RuleEvents; - fn iter_rules<'a>(&'a self) -> Box + 'a> { - Box::new(std::iter::empty()) - } - } - Self::new(Arc::new(EmptyRules) as Arc) } } + +/// A minimal empty rules container used for empty snapshots and tests. +#[derive(Debug)] +struct EmptyRules; + +impl crate::rules::subscriber_snapshot::RulesContainer for EmptyRules { + type Rule = dyn crate::rules::subscriber_snapshot::RuleEvents; + fn iter_rules<'a>(&'a self) -> Box + 'a> { + Box::new(std::iter::empty()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_s3_types::EventName; + + fn snapshot_with_mask(mask: u64) -> BucketRulesSnapshot { + BucketRulesSnapshot { + event_mask: mask, + rules: Arc::new(EmptyRules) as Arc, + } + } + + #[test] + fn store_then_load_roundtrips_snapshot() { + let index = SubscriberIndex::default(); + assert!(!index.has_subscriber("bucket", &EventName::ObjectCreatedPut)); + + index.store_snapshot("bucket", snapshot_with_mask(EventName::ObjectCreatedPut.mask())); + assert!(index.has_subscriber("bucket", &EventName::ObjectCreatedPut)); + + index.clear_bucket("bucket"); + assert!(!index.has_subscriber("bucket", &EventName::ObjectCreatedPut)); + } + + /// Regression test for backlog#984 (subscriber_index TOCTOU): many tasks + /// concurrently perform the first write for the *same* new bucket. The old + /// `get()`-then-`insert()` path could have concurrent first-writers each build + /// a distinct cell and clobber one another in the map, orphaning a stored + /// snapshot. With the atomic `compute_if_absent` upsert every writer shares the + /// one winning cell, so the final snapshot is always a real, non-empty store — + /// never the discarded empty default. + #[tokio::test(flavor = "multi_thread", worker_threads = 8)] + async fn concurrent_first_writes_share_one_cell() { + let mask = EventName::ObjectCreatedPut.mask(); + for _round in 0..64 { + let index = Arc::new(SubscriberIndex::default()); + const WRITERS: usize = 16; + + let mut handles = Vec::with_capacity(WRITERS); + for _ in 0..WRITERS { + let index = index.clone(); + handles.push(tokio::spawn(async move { + index.store_snapshot("shared-bucket", snapshot_with_mask(mask)); + })); + } + for handle in handles { + handle.await.expect("writer task must not panic"); + } + + assert!( + index.has_subscriber("shared-bucket", &EventName::ObjectCreatedPut), + "a concurrently-stored snapshot must survive; none was lost to a clobbered cell" + ); + } + } +} diff --git a/crates/notify/src/rules/xml_config.rs b/crates/notify/src/rules/xml_config.rs index 680ee8774..f2f3fd7e1 100644 --- a/crates/notify/src/rules/xml_config.rs +++ b/crates/notify/src/rules/xml_config.rs @@ -70,13 +70,18 @@ impl FilterRule { return Err(ParseConfigError::InvalidFilterName(self.name.clone())); } // ValidateFilterRuleValue from Go: - // no "." or ".." path segments, <= 1024 chars, valid UTF-8, no '\'. + // no "." or ".." path segments, <= 1024 characters, no '\'. for segment in self.value.split('/') { if segment == "." || segment == ".." { return Err(ParseConfigError::InvalidFilterValue(self.value.clone())); } } - if self.value.len() > 1024 || self.value.contains('\\') || std::str::from_utf8(self.value.as_bytes()).is_err() { + // The limit is 1024 *characters* (runes), matching S3/Go semantics. Using the + // byte length (`str::len`) would wrongly reject valid keys whose multi-byte + // UTF-8 encoding exceeds 1024 bytes while staying under 1024 characters + // (backlog#984). `self.value` is a `String`, so it is already valid UTF-8 — + // no separate UTF-8 check is needed. + if self.value.chars().count() > 1024 || self.value.contains('\\') { return Err(ParseConfigError::InvalidFilterValue(self.value.clone())); } Ok(()) @@ -374,3 +379,43 @@ impl NotificationConfiguration { // You may also need to set the default value here. But according to the current definition, they only contain ARN strings. } } + +#[cfg(test)] +mod filter_rule_length_tests { + use super::{FilterRule, ParseConfigError}; + + fn prefix(value: String) -> FilterRule { + FilterRule { + name: "prefix".to_string(), + value, + } + } + + #[test] + fn length_limit_counts_characters_not_bytes() { + // 1024 multi-byte characters: 3072 bytes but exactly 1024 chars — must pass. + // The old byte-length check (`str::len`) would have rejected this valid key + // (backlog#984). + let multibyte = "中".repeat(1024); + assert_eq!(multibyte.chars().count(), 1024); + assert!(multibyte.len() > 1024, "test setup: byte length must exceed the char limit"); + prefix(multibyte).validate().expect("1024-character value must be accepted"); + + // 1025 characters must be rejected. + let too_long = "a".repeat(1025); + assert!(matches!(prefix(too_long).validate(), Err(ParseConfigError::InvalidFilterValue(_)))); + + // Exactly 1024 ASCII characters must pass. + prefix("a".repeat(1024)) + .validate() + .expect("1024 ASCII chars must be accepted"); + } + + #[test] + fn backslash_is_still_rejected() { + assert!(matches!( + prefix("bad\\path".to_string()).validate(), + Err(ParseConfigError::InvalidFilterValue(_)) + )); + } +}