From dbc628f1699647c5dcd97ce8bc322a764fa5a7d8 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 8 Jul 2026 15:02:05 +0800 Subject: [PATCH] fix(audit): propagate dispatch delivery failures instead of swallowing them (#4424) fix(audit): propagate dispatch delivery failures instead of swallowing them (backlog#962) AuditPipeline::dispatch and dispatch_batch accumulated per-target save() errors, logged them, and then unconditionally returned Ok(()). When every configured target failed the audit entry was lost outright (for store-backed targets a failed save() means the event was neither delivered nor persisted for replay), yet callers such as dispatch_audit_log saw success and assumed the audit trail was intact. Audit is a compliance-critical path, so a total delivery failure that reports success is a silent data-loss bug. Both methods now distinguish three outcomes: all targets succeeded (Ok), partial failure where at least one target accepted the event (log a warning and return Ok, since the entry is not lost), and total failure where no delivery succeeded while errors were recorded (record the failure metric, log an error, and return Err(AuditError::Target) carrying the first target error). This lets the caller react instead of assuming success. Adds regression tests with a FailingTarget mock asserting that dispatch and dispatch_batch return Err on total failure and Ok on partial failure. --- crates/audit/src/pipeline.rs | 57 ++++++++- crates/audit/tests/pipeline_layer_test.rs | 136 +++++++++++++++++++++- 2 files changed, 185 insertions(+), 8 deletions(-) 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()));