From 2c7030d55e9b3979dd3f80c3209665a4d67483c2 Mon Sep 17 00:00:00 2001 From: mxsm Date: Tue, 7 Jul 2026 09:53:59 +0800 Subject: [PATCH 1/4] =?UTF-8?q?[ISSUE=20#8102]=F0=9F=90=9BPreserve=20ACL?= =?UTF-8?q?=20permission=20denied=20semantics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- rocketmq-auth/src/authorization/provider.rs | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/rocketmq-auth/src/authorization/provider.rs b/rocketmq-auth/src/authorization/provider.rs index 5808376cd..ee0c8da64 100644 --- a/rocketmq-auth/src/authorization/provider.rs +++ b/rocketmq-auth/src/authorization/provider.rs @@ -111,9 +111,7 @@ impl From for RocketMQError { fn from(error: AuthorizationError) -> Self { let message = error.to_string(); match error { - AuthorizationError::PermissionDenied { .. } => { - RocketMQError::Authentication(AuthError::AuthorizationFailed(message)) - } + AuthorizationError::PermissionDenied { .. } => RocketMQError::BrokerPermissionDenied { operation: message }, AuthorizationError::SubjectNotFound(_) | AuthorizationError::ResourceNotFound(_) | AuthorizationError::InvalidContext(_) => RocketMQError::illegal_argument(message), @@ -693,10 +691,7 @@ mod tests { resource: "topic:test".to_string(), reason: "insufficient permissions".to_string(), }); - assert!(matches!( - denied, - RocketMQError::Authentication(AuthError::AuthorizationFailed(_)) - )); + assert!(matches!(denied, RocketMQError::BrokerPermissionDenied { .. })); let invalid = RocketMQError::from(AuthorizationError::InvalidContext("missing subject".to_string())); assert!(matches!(invalid, RocketMQError::IllegalArgument(_))); From e7accfe06fbe0cc5108420ec0ef4e07aecd278d2 Mon Sep 17 00:00:00 2001 From: mxsm Date: Tue, 7 Jul 2026 10:40:34 +0800 Subject: [PATCH 2/4] =?UTF-8?q?[ISSUE=20#8102]=F0=9F=90=9BFix=20observabil?= =?UTF-8?q?ity=20warnings=20as=20errors?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- rocketmq-observability/src/init.rs | 3 +++ rocketmq-observability/src/logging.rs | 6 +++--- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/rocketmq-observability/src/init.rs b/rocketmq-observability/src/init.rs index 52870ebf9..6c73d0418 100644 --- a/rocketmq-observability/src/init.rs +++ b/rocketmq-observability/src/init.rs @@ -201,7 +201,10 @@ pub(crate) fn init_telemetry_providers(config: &ObservabilityConfig) -> Result Result<(Vec, LoggingGuard), ObservabilityError> { let mut layers = Vec::new(); let mut logging_guard = LoggingGuard::noop(); @@ -168,7 +168,7 @@ fn build_subscriber_layers( } #[cfg(feature = "otel-traces")] - if let Some(tracer_provider) = telemetry_guard.tracer_provider() { + if let Some(tracer_provider) = _telemetry_guard.tracer_provider() { layers.push(Box::new(crate::trace::build_tracing_layer( &config.observability, tracer_provider, @@ -176,7 +176,7 @@ fn build_subscriber_layers( } #[cfg(feature = "otel-logs")] - if let Some(logger_provider) = telemetry_guard.logger_provider() { + if let Some(logger_provider) = _telemetry_guard.logger_provider() { layers.push(Box::new(crate::logs::bridge::build_logs_layer(logger_provider))); } From 2147bd02b820258e4e245733d48a16725e480b0a Mon Sep 17 00:00:00 2001 From: mxsm Date: Tue, 7 Jul 2026 14:33:28 +0800 Subject: [PATCH 3/4] =?UTF-8?q?[ISSUE=20#8102]=F0=9F=90=9BStabilize=20auth?= =?UTF-8?q?=20ACL=20watcher=20probe?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- rocketmq-auth/src/lib.rs | 51 ++++++++++++++++++++++++++++++---------- 1 file changed, 39 insertions(+), 12 deletions(-) diff --git a/rocketmq-auth/src/lib.rs b/rocketmq-auth/src/lib.rs index 3fba68166..3636cda1c 100644 --- a/rocketmq-auth/src/lib.rs +++ b/rocketmq-auth/src/lib.rs @@ -142,9 +142,15 @@ pub mod bench_support { let deadline = Instant::now() + Duration::from_secs(2); let reload_success = loop { - let user = authn_provider.get_user("alice").await?; - if user.password().map(|value| value.as_str()) == Some("second") { - break true; + match authn_provider.get_user("alice").await { + Ok(user) if user.password().map(|value| value.as_str()) == Some("second") => { + break true; + } + Ok(_) => {} + Err(rocketmq_error::RocketMQError::Authentication(rocketmq_error::AuthError::UserNotFound( + username, + ))) if username == "alice" => {} + Err(error) => return Err(error), } if Instant::now() >= deadline { break false; @@ -294,19 +300,40 @@ pub mod bench_support { } fn write_acl_file(path: &std::path::Path, secret: &str) -> RocketMQResult<()> { - fs::write( - path, - format!( - r#" + let content = format!( + r#" accounts: - accessKey: alice secretKey: {secret} "# - ), - ) - .map_err(|error| { - rocketmq_error::RocketMQError::storage_write_failed(path.display().to_string(), error.to_string()) - }) + ); + let temp_file = temp_acl_file_path(path); + fs::write(&temp_file, content).map_err(|error| { + rocketmq_error::RocketMQError::storage_write_failed(temp_file.display().to_string(), error.to_string()) + })?; + + match fs::rename(&temp_file, path) { + Ok(()) => Ok(()), + Err(rename_error) => { + fs::copy(&temp_file, path).map_err(|error| { + rocketmq_error::RocketMQError::storage_write_failed( + path.display().to_string(), + format!("{error}; rename failed first: {rename_error}"), + ) + })?; + let _ = fs::remove_file(&temp_file); + Ok(()) + } + } + } + + fn temp_acl_file_path(path: &std::path::Path) -> PathBuf { + let file_name = path + .file_name() + .and_then(|value| value.to_str()) + .unwrap_or("plain_acl.yml"); + let id = NEXT_ACL_WATCHER_PROBE_ID.fetch_add(1, Ordering::Relaxed); + path.with_file_name(format!(".{file_name}.{id}.tmp")) } impl From for AuthSyncBridgeCounterSnapshot { From cce538c8897348d8c8fa6a7f5487c542c54ddaab Mon Sep 17 00:00:00 2001 From: mxsm Date: Tue, 7 Jul 2026 15:07:14 +0800 Subject: [PATCH 4/4] =?UTF-8?q?[ISSUE=20#8102]=F0=9F=90=9BKeep=20ACL=20wat?= =?UTF-8?q?cher=20probe=20coverage=20focused?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- rocketmq-auth/src/lib.rs | 30 +++++++++++++++++------------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/rocketmq-auth/src/lib.rs b/rocketmq-auth/src/lib.rs index 3636cda1c..60bdef513 100644 --- a/rocketmq-auth/src/lib.rs +++ b/rocketmq-auth/src/lib.rs @@ -312,19 +312,7 @@ accounts: rocketmq_error::RocketMQError::storage_write_failed(temp_file.display().to_string(), error.to_string()) })?; - match fs::rename(&temp_file, path) { - Ok(()) => Ok(()), - Err(rename_error) => { - fs::copy(&temp_file, path).map_err(|error| { - rocketmq_error::RocketMQError::storage_write_failed( - path.display().to_string(), - format!("{error}; rename failed first: {rename_error}"), - ) - })?; - let _ = fs::remove_file(&temp_file); - Ok(()) - } - } + replace_acl_file(&temp_file, path) } fn temp_acl_file_path(path: &std::path::Path) -> PathBuf { @@ -336,6 +324,22 @@ accounts: path.with_file_name(format!(".{file_name}.{id}.tmp")) } + #[cfg(not(windows))] + fn replace_acl_file(temp_file: &std::path::Path, path: &std::path::Path) -> RocketMQResult<()> { + fs::rename(temp_file, path).map_err(|error| { + rocketmq_error::RocketMQError::storage_write_failed(path.display().to_string(), error.to_string()) + }) + } + + #[cfg(windows)] + fn replace_acl_file(temp_file: &std::path::Path, path: &std::path::Path) -> RocketMQResult<()> { + fs::copy(temp_file, path).map_err(|error| { + rocketmq_error::RocketMQError::storage_write_failed(path.display().to_string(), error.to_string()) + })?; + let _ = fs::remove_file(temp_file); + Ok(()) + } + impl From for AuthSyncBridgeCounterSnapshot { fn from(snapshot: crate::runtime_bridge::AuthSyncBridgeSnapshot) -> Self { Self {