diff --git a/Cargo.lock b/Cargo.lock index 4fec6d622..8d14008e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9536,7 +9536,6 @@ dependencies = [ "tracing", "tracing-subscriber", "url", - "wildmatch", ] [[package]] @@ -12454,15 +12453,6 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72069c3113ab32ab29e5584db3c6ec55d416895e60715417b5b883a357c3e471" -[[package]] -name = "wildmatch" -version = "2.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29333c3ea1ba8b17211763463ff24ee84e41c78224c16b001cd907e663a38c68" -dependencies = [ - "serde", -] - [[package]] name = "winapi" version = "0.3.9" diff --git a/crates/notify/Cargo.toml b/crates/notify/Cargo.toml index efa9ba475..1508c8064 100644 --- a/crates/notify/Cargo.toml +++ b/crates/notify/Cargo.toml @@ -46,7 +46,6 @@ thiserror = { workspace = true } tokio = { workspace = true, features = ["rt-multi-thread", "sync", "time"] } tracing = { workspace = true } url = { workspace = true } -wildmatch = { workspace = true, features = ["serde"] } metrics = { workspace = true } # quick-xml dependencies for custom S3KeyFilter XML deserialization diff --git a/crates/notify/src/notifier.rs b/crates/notify/src/notifier.rs index ba71b9698..601a50dad 100644 --- a/crates/notify/src/notifier.rs +++ b/crates/notify/src/notifier.rs @@ -206,9 +206,18 @@ impl EventNotifier { "Failed to dispatch notify event" ); } else { - if is_deferred { - metrics.decrement_processing(); - } else { + // Counting lifecycle (backlog#979): `increment_processing` above marks + // the event as in-flight exactly once. For a non-deferred target, `save` + // performs synchronous delivery, so the event is fully processed here and + // we transition it from processing to processed. For a deferred + // (store-backed) target, `save` only enqueues the event to the store: it + // stays in-flight until the replay worker actually delivers it and emits + // `ReplayEvent::Delivered`, which is the single place that calls + // `increment_processed` for that event. Decrementing `processing_events` + // here as well would double-count the completion and underflow the + // counter back to `usize::MAX`, corrupting metrics and any in-flight + // backpressure decisions. + if !is_deferred { metrics.increment_processed(); } debug!( @@ -417,7 +426,7 @@ mod tests { use rustfs_targets::StoreError; use rustfs_targets::{ TargetError, - store::{Key, Store}, + store::{Key, QueueStore, Store}, target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}, }; use std::sync::{ @@ -609,4 +618,113 @@ mod tests { assert_eq!(target.save_calls.load(Ordering::SeqCst), 1); } + + /// A store-backed (deferred) target. `save` only enqueues to the store, so + /// the actual delivery happens later in the replay worker. + #[derive(Clone)] + struct DeferredTestTarget { + id: TargetID, + save_calls: Arc, + store: QueueStore, + } + + impl DeferredTestTarget { + fn new(id: &str, name: &str) -> Self { + Self { + id: TargetID::new(id.to_string(), name.to_string()), + save_calls: Arc::new(AtomicUsize::new(0)), + // The store is never actually written to here: `save` below only bumps a + // counter. It just has to exist so the notifier treats this target as + // store-backed (deferred delivery), exercising the deferred counting path. + store: QueueStore::new(std::env::temp_dir().join("rustfs-notify-979-noop-store"), 0, ""), + } + } + } + + #[async_trait] + impl Target for DeferredTestTarget + 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> { + self.save_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + + async fn send_raw_from_store(&self, _key: Key, _body: Vec, _meta: QueuedPayloadMeta) -> Result<(), TargetError> { + Ok(()) + } + + async fn close(&self) -> Result<(), TargetError> { + Ok(()) + } + + fn store(&self) -> Option<&(dyn Store + Send + Sync)> { + Some(&self.store) + } + + fn clone_dyn(&self) -> Box + Send + Sync> { + Box::new(self.clone()) + } + + async fn init(&self) -> Result<(), TargetError> { + Ok(()) + } + + fn is_enabled(&self) -> bool { + true + } + } + + /// Regression test for backlog#979 (a): dispatching to a store-backed + /// (deferred) target must count the event as in-flight exactly once and must + /// not decrement `processing_events` at enqueue time. The replay worker is the + /// single owner of the completion decrement (`increment_processed`); an extra + /// decrement here previously double-counted and underflowed the counter back + /// to `usize::MAX`. + #[tokio::test] + async fn deferred_target_dispatch_does_not_underflow_processing_counter() { + let metrics = Arc::new(NotificationMetrics::new()); + let rule_engine = NotifyRuleEngine::new(); + let notifier = EventNotifier::new(metrics.clone(), rule_engine.clone()); + + let target = DeferredTestTarget::new("deferred-target", "webhook"); + let mut rules_map = RulesMap::new(); + rules_map.add_rule_config(&[EventName::ObjectCreatedPut], "*".to_string(), target.id.clone()); + rule_engine.set_bucket_rules("bucket", rules_map).await; + notifier.target_list().write().await.add(Arc::new(target.clone())).unwrap(); + + // Dispatch enqueues to the store; the event stays in-flight (processing == 1). + notifier + .send(Arc::new(Event::new_test_event("bucket", "object", EventName::ObjectCreatedPut))) + .await; + + assert_eq!(target.save_calls.load(Ordering::SeqCst), 1); + assert_eq!( + metrics.processing_count(), + 1, + "a queued deferred event must remain counted as in-flight after enqueue" + ); + assert_eq!(metrics.processed_count(), 0); + + // Simulate the replay worker delivering the queued event (runtime_facade + // maps `ReplayEvent::Delivered` to `increment_processed`). + metrics.increment_processed(); + + assert_eq!( + metrics.processing_count(), + 0, + "processing counter must return to zero, not underflow to usize::MAX" + ); + assert_ne!(metrics.processing_count(), usize::MAX); + assert_eq!(metrics.processed_count(), 1); + } } diff --git a/crates/notify/src/rules/pattern.rs b/crates/notify/src/rules/pattern.rs index 34e5e05e3..8bd4fdfeb 100644 --- a/crates/notify/src/rules/pattern.rs +++ b/crates/notify/src/rules/pattern.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -use wildmatch::WildMatch; - /// Create new pattern string based on prefix and suffix。 /// /// The rule is similar to event.NewPattern in the Go version: @@ -71,7 +69,54 @@ pub fn match_simple(pattern_str: &str, object_name: &str) -> bool { return false; // Or true if an empty pattern means "match all" in some contexts. // Given Go's NewRulesMap defaults to "*", an empty pattern from Filter is unlikely to mean "match all". } - WildMatch::new(pattern_str).matches(object_name) + glob_match_star_only(pattern_str, object_name) +} + +/// Matches `object_name` against a glob `pattern` using **S3 filter semantics**: +/// only `*` is a wildcard (matching any run of characters, including none), and +/// every other character — crucially including `?` — is matched literally. +/// +/// S3 event notification prefix/suffix filters treat `*` as the only wildcard; +/// AWS does not assign any special meaning to `?`. Earlier this matcher delegated +/// to `wildmatch`, which interprets `?` as a single-character wildcard, so a +/// literal `?` in a prefix/suffix filter (or object key) would over-match and +/// misroute events (backlog#979). This hand-rolled matcher keeps `?` literal. +fn glob_match_star_only(pattern: &str, text: &str) -> bool { + let pattern: Vec = pattern.chars().collect(); + let text: Vec = text.chars().collect(); + + let mut p = 0usize; // cursor into `pattern` + let mut t = 0usize; // cursor into `text` + // Backtracking state: the last `*` we saw and where in `text` we matched it. + let mut star_at: Option = None; + let mut star_match_t = 0usize; + + while t < text.len() { + if p < pattern.len() && pattern[p] == '*' { + // Record the star and tentatively let it consume nothing. + star_at = Some(p); + star_match_t = t; + p += 1; + } else if p < pattern.len() && pattern[p] == text[t] { + // Literal match (any non-`*` char, including `?`). + p += 1; + t += 1; + } else if let Some(star_p) = star_at { + // Mismatch: let the previous `*` absorb one more character and retry. + p = star_p + 1; + star_match_t += 1; + t = star_match_t; + } else { + return false; + } + } + + // Consume any trailing `*` in the pattern. + while p < pattern.len() && pattern[p] == '*' { + p += 1; + } + + p == pattern.len() } #[cfg(test)] @@ -108,4 +153,31 @@ mod tests { assert!(match_simple("a*b*c", "abc")); assert!(match_simple("a*b*c", "axbc")); } + + /// Regression test for backlog#979 (c): S3 filter semantics treat `*` as the + /// only wildcard. `?` must be matched literally, not as a single-character + /// wildcard (which is how the previous `wildmatch`-based matcher behaved). + #[test] + fn question_mark_is_literal_not_single_char_wildcard() { + // `?` is a literal: it only matches a literal `?`, never an arbitrary char. + assert!(!match_simple("a?c", "abc")); + assert!(match_simple("a?c", "a?c")); + + // Prefix-derived pattern containing a literal `?`. + let prefix_pattern = new_pattern(Some("foo?"), None); + assert_eq!(prefix_pattern, "foo?*"); + assert!(!match_simple(&prefix_pattern, "fooX/bar")); // `?` must not match `X` + assert!(match_simple(&prefix_pattern, "foo?bar")); // literal `?` matches + assert!(!match_simple(&prefix_pattern, "foobar")); // `?` is required literally + + // Suffix-derived pattern containing a literal `?`. + let suffix_pattern = new_pattern(None, Some("?.log")); + assert_eq!(suffix_pattern, "*?.log"); + assert!(match_simple(&suffix_pattern, "app?.log")); + assert!(!match_simple(&suffix_pattern, "appX.log")); + + // `*` continues to work as a multi-character wildcard alongside literal `?`. + assert!(match_simple("a?*c", "a?bbbc")); + assert!(!match_simple("a?*c", "aXbbbc")); + } }