diff --git a/crates/audit/src/pipeline.rs b/crates/audit/src/pipeline.rs index bdc53468e..df5f04fdd 100644 --- a/crates/audit/src/pipeline.rs +++ b/crates/audit/src/pipeline.rs @@ -114,19 +114,42 @@ impl AuditPipeline { if errors.is_empty() { observability::record_audit_success(dispatch_time); - } else { - observability::record_audit_failure(dispatch_time); - warn!( + return Ok(()); + } + + observability::record_audit_failure(dispatch_time); + let error_count = errors.len(); + + if success_count == 0 { + // Every configured target rejected the event. For store-backed targets a + // failed save() means the entry was neither delivered nor persisted for + // replay, so it is lost outright. Propagate the failure instead of + // returning Ok so the caller can react (alert, degrade, or reject the + // request) rather than assume the audit trail is intact. + error!( event = EVENT_AUDIT_DISPATCH_FAILED, component = LOG_COMPONENT_AUDIT, subsystem = LOG_SUBSYSTEM_PIPELINE, - error_count = errors.len(), - success_count = success_count, + error_count = error_count, duration_ms = dispatch_time.as_millis() as u64, - "Some audit targets failed to receive audit event" + "All audit targets failed to receive audit event" ); + // `errors` is non-empty here, so `remove(0)` cannot panic. + return Err(crate::AuditError::Target(errors.remove(0))); } + // Partial failure: at least one target accepted the event, so the entry is + // not lost. Surface the degradation but let the dispatch succeed. + warn!( + event = EVENT_AUDIT_DISPATCH_FAILED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + error_count = error_count, + success_count = success_count, + duration_ms = dispatch_time.as_millis() as u64, + "Some audit targets failed to receive audit event" + ); + Ok(()) } @@ -179,6 +202,7 @@ impl AuditPipeline { let results = futures::future::join_all(tasks).await; let mut total_success = 0; let mut total_errors = 0; + let mut first_error: Option = None; for (target_id, success_count, errors) in results { total_success += success_count; total_errors += errors.len(); @@ -191,6 +215,9 @@ impl AuditPipeline { error = ?e, "Audit batch dispatch failed" ); + if first_error.is_none() { + first_error = Some(e); + } } } @@ -206,6 +233,24 @@ impl AuditPipeline { "Completed audit batch dispatch" ); + // No save() across any target/entry succeeded while errors were recorded: + // the batch was lost entirely. Propagate rather than silently returning Ok. + if total_errors > 0 && total_success == 0 { + observability::record_audit_failure(dispatch_time); + error!( + event = EVENT_AUDIT_BATCH_DISPATCH_FAILED, + component = LOG_COMPONENT_AUDIT, + subsystem = LOG_SUBSYSTEM_PIPELINE, + entry_count = entries.len(), + error_count = total_errors, + duration_ms = dispatch_time.as_millis() as u64, + "All audit targets failed to receive audit batch" + ); + return Err(crate::AuditError::Target( + first_error.expect("total_errors > 0 guarantees a captured target error"), + )); + } + Ok(()) } diff --git a/crates/audit/tests/pipeline_layer_test.rs b/crates/audit/tests/pipeline_layer_test.rs index 8e76b1081..3b28b5863 100644 --- a/crates/audit/tests/pipeline_layer_test.rs +++ b/crates/audit/tests/pipeline_layer_test.rs @@ -13,11 +13,11 @@ // limitations under the License. use async_trait::async_trait; -use rustfs_audit::{AuditPipeline, AuditRegistry, AuditRuntimeFacade, AuditRuntimeView}; +use rustfs_audit::{AuditEntry, AuditError, AuditPipeline, AuditRegistry, AuditRuntimeFacade, AuditRuntimeView}; use rustfs_targets::arn::TargetID; use rustfs_targets::store::{Key, Store}; use rustfs_targets::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta}; -use rustfs_targets::{StoreError, Target, TargetError}; +use rustfs_targets::{SharedTarget, StoreError, Target, TargetError}; use serde::{Serialize, de::DeserializeOwned}; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -84,6 +84,138 @@ where } } +/// A target whose `save()` always fails, used to exercise the dispatch +/// failure-propagation paths. +#[derive(Clone)] +struct FailingTarget { + id: TargetID, + save_calls: Arc, +} + +impl FailingTarget { + fn new(id: &str, name: &str) -> Self { + Self { + id: TargetID::new(id.to_string(), name.to_string()), + save_calls: Arc::new(AtomicUsize::new(0)), + } + } +} + +#[async_trait] +impl Target for FailingTarget +where + E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned, +{ + 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); + Err(TargetError::Storage("disk full".to_string())) + } + + 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)> { + 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 + } +} + +fn pipeline_with_targets(targets: Vec<(&str, SharedTarget)>) -> AuditPipeline { + let mut registry = AuditRegistry::new(); + for (id, target) in targets { + registry.add_shared_target(id.to_string(), target); + } + AuditPipeline::new(Arc::new(Mutex::new(registry))) +} + +#[tokio::test] +async fn audit_pipeline_dispatch_propagates_total_failure() { + let failing = FailingTarget::new("primary", "webhook"); + let save_calls = Arc::clone(&failing.save_calls); + let pipeline = pipeline_with_targets(vec![("primary:webhook", Arc::new(failing))]); + + let result = pipeline.dispatch(Arc::new(AuditEntry::default())).await; + + assert!( + matches!(result, Err(AuditError::Target(_))), + "dispatch must surface an error when every target fails, got {result:?}" + ); + assert_eq!(save_calls.load(Ordering::SeqCst), 1, "the failing target should have been invoked"); +} + +#[tokio::test] +async fn audit_pipeline_dispatch_tolerates_partial_failure() { + let failing = FailingTarget::new("primary", "webhook"); + let healthy = TestTarget::new("secondary", "webhook"); + let pipeline = pipeline_with_targets(vec![ + ("primary:webhook", Arc::new(failing)), + ("secondary:webhook", Arc::new(healthy)), + ]); + + let result = pipeline.dispatch(Arc::new(AuditEntry::default())).await; + + assert!( + result.is_ok(), + "dispatch should succeed when at least one target accepts the event, got {result:?}" + ); +} + +#[tokio::test] +async fn audit_pipeline_dispatch_batch_propagates_total_failure() { + let failing = FailingTarget::new("primary", "webhook"); + let pipeline = pipeline_with_targets(vec![("primary:webhook", Arc::new(failing))]); + + let entries = vec![Arc::new(AuditEntry::default()), Arc::new(AuditEntry::default())]; + let result = pipeline.dispatch_batch(entries).await; + + assert!( + matches!(result, Err(AuditError::Target(_))), + "dispatch_batch must surface an error when every delivery fails, got {result:?}" + ); +} + +#[tokio::test] +async fn audit_pipeline_dispatch_batch_tolerates_partial_failure() { + let failing = FailingTarget::new("primary", "webhook"); + let healthy = TestTarget::new("secondary", "webhook"); + let pipeline = pipeline_with_targets(vec![ + ("primary:webhook", Arc::new(failing)), + ("secondary:webhook", Arc::new(healthy)), + ]); + + let entries = vec![Arc::new(AuditEntry::default())]; + let result = pipeline.dispatch_batch(entries).await; + + assert!( + result.is_ok(), + "dispatch_batch should succeed when a healthy target accepts the batch, got {result:?}" + ); +} + #[tokio::test] async fn audit_runtime_view_lists_empty_targets() { let registry = Arc::new(Mutex::new(AuditRegistry::new()));