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(())
     }