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(_))); diff --git a/rocketmq-auth/src/lib.rs b/rocketmq-auth/src/lib.rs index 3fba68166..60bdef513 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,21 +300,46 @@ 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| { + ); + 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()) + })?; + + replace_acl_file(&temp_file, path) + } + + 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")) + } + + #[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 { 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))); }