From c1bcee327c3ba0329722e32f1528c01f5823799f Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Thu, 14 May 2026 01:10:14 +0800 Subject: [PATCH] fix(notify): keep live listen events active when disabled (#2952) Co-authored-by: houseme --- crates/e2e_test/src/object_lambda_test.rs | 35 +++++++++++++++++++++++ crates/notify/src/global.rs | 14 +++++++++ crates/notify/src/lib.rs | 2 +- rustfs/src/server/event.rs | 14 ++++++++- rustfs/src/storage/helper.rs | 6 +++- 5 files changed, 68 insertions(+), 3 deletions(-) diff --git a/crates/e2e_test/src/object_lambda_test.rs b/crates/e2e_test/src/object_lambda_test.rs index 0e35f4e5d..897a4636b 100644 --- a/crates/e2e_test/src/object_lambda_test.rs +++ b/crates/e2e_test/src/object_lambda_test.rs @@ -1150,6 +1150,41 @@ async fn test_listen_notification_emits_after_put_object() -> Result<(), Box Result<(), Box> { + init_logging(); + + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server_with_env(vec![], &[("RUSTFS_NOTIFY_ENABLE", "false")]) + .await?; + + let bucket = "listen-empty-bucket-e2e"; + let key = "seed/object.txt"; + let client = env.create_s3_client(); + + client.create_bucket().bucket(bucket).send().await?; + + let listen_url = format!("{}/{bucket}?events={}&ping=1", env.url, urlencoding::encode("s3:ObjectCreated:*"),); + let response = signed_request(http::Method::GET, &listen_url, &env.access_key, &env.secret_key, None, None).await?; + assert_eq!(response.status(), StatusCode::OK); + + let read_task = tokio::spawn(read_listen_notification_event(response, key)); + + client + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from_static(b"empty bucket watch body")) + .send() + .await?; + + let payload = timeout(Duration::from_secs(12), read_task).await???; + assert!(!payload.is_empty(), "listen_notification payload should not be empty"); + + Ok(()) +} + #[tokio::test] #[serial] async fn test_listen_notification_fans_in_remote_node_events() -> Result<(), Box> { diff --git a/crates/notify/src/global.rs b/crates/notify/src/global.rs index a0c716f7e..767ce8e76 100644 --- a/crates/notify/src/global.rs +++ b/crates/notify/src/global.rs @@ -38,6 +38,20 @@ pub async fn initialize(config: Config) -> Result<(), NotificationError> { } } +/// Initialize the global notification system only for live in-process consumers. +/// +/// This does not load configured notification targets or bucket rules. It exists so +/// ListenBucketNotification clients can receive live events even when external +/// notification targets are disabled. +pub fn initialize_live_events() -> Result<(), NotificationError> { + let system = NotificationSystem::new(Config::new()); + + match NOTIFICATION_SYSTEM.set(Arc::new(system)) { + Ok(_) => Ok(()), + Err(_) => Err(NotificationError::Lifecycle(LifecycleError::AlreadyInitialized)), + } +} + /// Returns a handle to the global NotificationSystem instance. /// Return None if the system has not been initialized. pub fn notification_system() -> Option> { diff --git a/crates/notify/src/lib.rs b/crates/notify/src/lib.rs index 267367230..298f7feec 100644 --- a/crates/notify/src/lib.rs +++ b/crates/notify/src/lib.rs @@ -32,7 +32,7 @@ pub mod stream; pub use error::{LifecycleError, NotificationError}; pub use event::{Event, EventArgs, EventArgsBuilder}; pub use global::{ - initialize, is_notification_system_initialized, notification_metrics_snapshot, notification_system, + initialize, initialize_live_events, is_notification_system_initialized, notification_metrics_snapshot, notification_system, notification_target_metrics, notifier_global, }; pub use integration::{NotificationMetricSnapshot, NotificationSystem, NotificationTargetMetricSnapshot}; diff --git a/rustfs/src/server/event.rs b/rustfs/src/server/event.rs index 10bf50f24..1de42c696 100644 --- a/rustfs/src/server/event.rs +++ b/rustfs/src/server/event.rs @@ -107,9 +107,21 @@ pub async fn init_event_notifier() { if !enabled { info!( target: "rustfs::main::init_event_notifier", - "Notify module is disabled, event notifier initialization is skipped. Set {}=true to enable notify initialization.", + "Notify module is disabled, initializing live event stream support only. Set {}=true to enable notification targets.", rustfs_config::ENV_NOTIFY_ENABLE ); + if rustfs_notify::notification_system().is_none() { + match rustfs_notify::initialize_live_events() { + Ok(()) => { + install_ecstore_event_dispatch_hook(); + info!( + target: "rustfs::main::init_event_notifier", + "Live event stream support initialized successfully." + ); + } + Err(e) => error!("Failed to initialize live event stream support: {}", e), + } + } return; } diff --git a/rustfs/src/storage/helper.rs b/rustfs/src/storage/helper.rs index 5d73328be..5f85a3169 100644 --- a/rustfs/src/storage/helper.rs +++ b/rustfs/src/storage/helper.rs @@ -90,7 +90,7 @@ impl OperationHelper { /// Create a new OperationHelper for S3 requests. pub fn new(req: &S3Request, event: EventName, op: S3Operation) -> Self { let audit_enabled = is_audit_module_enabled(); - let notify_enabled = is_notify_module_enabled(); + let notify_enabled = should_build_notification_event(is_notify_module_enabled()); let path = req.uri.path().trim_start_matches('/'); let mut segs = path.splitn(2, '/'); @@ -334,6 +334,10 @@ impl OperationHelper { } } +fn should_build_notification_event(notify_module_enabled: bool) -> bool { + notify_module_enabled || rustfs_notify::notification_system().is_some_and(|system| system.has_live_listeners()) +} + impl Drop for OperationHelper { fn drop(&mut self) { let Self::Enabled(state) = self else {