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

@@ -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<rustfs_targets::TargetError> = 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(())
     }
 

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<AtomicUsize>,
+}
+
+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<E> Target<E> for FailingTarget
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    fn id(&self) -> TargetID {
+        self.id.clone()
+    }
+
+    async fn is_active(&self) -> Result<bool, TargetError> {
+        Ok(true)
+    }
+
+    async fn save(&self, _event: Arc<EntityTarget<E>>) -> 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<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
+        Ok(())
+    }
+
+    async fn close(&self) -> Result<(), TargetError> {
+        Ok(())
+    }
+
+    fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
+        None
+    }
+
+    fn clone_dyn(&self) -> Box<dyn Target<E> + 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<AuditEntry>)>) -> 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()));