fix(audit): harden reload ordering, start race, paused drops, and batch observability (#4497)

Follow-up hardening for the audit control plane. The ABBA deadlock
(backlog#961), dispatch failure propagation (backlog#962), and credential
header redaction (backlog#963) already landed on main; this change addresses
the remaining audit-side findings.

- backlog#970: reload/commit now shuts down existing replay workers and closes
  old targets *before* activating the replacement set, so old and new workers
  never drain the same store concurrently and re-deliver entries. Extracted a
  state-neutral `shutdown_runtime_targets` helper (registry-then-cancellers
  lock order preserved).
- backlog#978: `start()` claims the `Starting` transition atomically under the
  state lock, closing the check-then-act race that could double-activate;
  `dispatch()` no longer returns Ok while paused, it surfaces an explicit
  `AuditError::Paused` so the audit trail is not silently corrupted.
- backlog#984 (audit part): `dispatch_audit_log` performs a single state read
  and interprets not-running/paused as a deliberate skip while surfacing real
  delivery failures; `dispatch_batch` records the same observability signals as
  single dispatch; documented the unordered cross-target fan-out.

Adds unit tests: paused dispatch returns Err, concurrent start does not hang or
double-activate, commit closes old targets before installing new, and
dispatch/dispatch_batch delivery-outcome coverage (all-fail -> Err, partial ->
Ok, all-success -> Ok).

Relates to rustfs/backlog#970
Relates to rustfs/backlog#978
Relates to rustfs/backlog#984

Co-authored-by: heihutu <heihutu@gmail.com>

crates/audit/src/error.rs

@@ -35,6 +35,9 @@ pub enum AuditError {
     #[error("System already initialized")]
     AlreadyInitialized,
 
+    #[error("Audit system is paused; entry was not accepted")]
+    Paused,
+
     #[error("Storage not available: {0}")]
     StorageNotAvailable(String),
 

crates/audit/src/global.rs

@@ -12,7 +12,7 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
-use crate::{AuditEntry, AuditResult, AuditSystem, system::AuditTargetMetricSnapshot};
+use crate::{AuditEntry, AuditError, AuditResult, AuditSystem, system::AuditTargetMetricSnapshot};
 use rustfs_config::server_config::Config;
 use std::sync::{Arc, OnceLock};
 use tracing::{debug, error, trace};
@@ -78,10 +78,27 @@ pub async fn resume_audit_system() -> AuditResult<()> {
 
 /// Dispatch an audit log entry to all targets
 pub async fn dispatch_audit_log(entry: Arc<AuditEntry>) -> AuditResult<()> {
-    if let Some(system) = audit_system() {
-        if system.is_running().await {
-            system.dispatch(entry).await
-        } else {
+    let Some(system) = audit_system() else {
+        debug!(
+            event = EVENT_AUDIT_ENTRY_DROPPED,
+            component = LOG_COMPONENT_AUDIT,
+            subsystem = LOG_SUBSYSTEM_GLOBAL,
+            reason = "system_not_initialized",
+            "Dropped audit entry"
+        );
+        return Ok(());
+    };
+
+    // Single state read (backlog#984): the previous code checked `is_running()`
+    // and then called `dispatch()`, which re-read the state. Between the two
+    // reads the system could transition (e.g. Running -> Stopping) and
+    // `dispatch()` would return an error the caller never expected. Let
+    // `dispatch()` be the single authority on the current state and interpret
+    // its "not accepting" errors as a deliberate skip, while still surfacing
+    // real delivery failures (backlog#962).
+    match system.dispatch(entry).await {
+        Ok(()) => Ok(()),
+        Err(AuditError::NotInitialized(_)) | Err(AuditError::Paused) => {
             trace!(
                 event = EVENT_AUDIT_ENTRY_DROPPED,
                 component = LOG_COMPONENT_AUDIT,
@@ -91,15 +108,7 @@ pub async fn dispatch_audit_log(entry: Arc<AuditEntry>) -> AuditResult<()> {
             );
             Ok(())
         }
-    } else {
-        debug!(
-            event = EVENT_AUDIT_ENTRY_DROPPED,
-            component = LOG_COMPONENT_AUDIT,
-            subsystem = LOG_SUBSYSTEM_GLOBAL,
-            reason = "system_not_initialized",
-            "Dropped audit entry"
-        );
-        Ok(())
+        Err(e) => Err(e),
     }
 }
 

crates/audit/src/pipeline.rs

@@ -45,6 +45,12 @@ impl AuditPipeline {
         Self { registry }
     }
 
+    /// Fans an audit entry out to every configured target concurrently.
+    ///
+    /// Delivery across targets is unordered: the per-target `save()` calls run
+    /// via `join_all` and may complete in any order. Ordering of entries within
+    /// a single target is preserved by that target's own store/queue, not by
+    /// this fan-out.
     pub async fn dispatch(&self, entry: Arc<AuditEntry>) -> AuditResult<()> {
         let start_time = std::time::Instant::now();
 
@@ -190,8 +196,14 @@ impl AuditPipeline {
                         data: (*entry).clone(),
                     };
                     match target.save(Arc::new(entity_target)).await {
-                        Ok(_) => success_count += 1,
-                        Err(e) => errors.push(e),
+                        Ok(_) => {
+                            success_count += 1;
+                            observability::record_target_success();
+                        }
+                        Err(e) => {
+                            observability::record_target_failure();
+                            errors.push(e);
+                        }
                     }
                 }
                 (target.id().to_string(), success_count, errors)
@@ -251,6 +263,12 @@ impl AuditPipeline {
             ));
         }
 
+        // Record the aggregate event outcome so batch dispatch reports the same
+        // observability signal as single dispatch (backlog#984): full success or
+        // partial failure both count as a delivered audit event here, since at
+        // least one target accepted every entry that reached this point.
+        observability::record_audit_success(dispatch_time);
+
         Ok(())
     }
 
@@ -541,3 +559,135 @@ impl AuditRuntimeFacade {
         self.runtime_adapter.stop_replay_workers(&mut replay_workers).await;
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::AuditPipeline;
+    use crate::{AuditEntry, AuditError, AuditRegistry};
+    use async_trait::async_trait;
+    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 std::sync::Arc;
+    use tokio::sync::Mutex;
+
+    /// Mock target whose `save()` outcome is fixed at construction so tests can
+    /// force full-success / full-failure / partial-failure fan-outs.
+    #[derive(Clone)]
+    struct MockTarget {
+        id: TargetID,
+        fail: bool,
+    }
+
+    impl MockTarget {
+        fn new(id: &str, fail: bool) -> Self {
+            Self {
+                id: TargetID::new(id.to_string(), "webhook".to_string()),
+                fail,
+            }
+        }
+    }
+
+    #[async_trait]
+    impl<E> Target<E> for MockTarget
+    where
+        E: rustfs_targets::PluginEvent,
+    {
+        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> {
+            if self.fail {
+                Err(TargetError::Configuration("forced save failure".to_string()))
+            } else {
+                Ok(())
+            }
+        }
+
+        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())
+        }
+
+        fn is_enabled(&self) -> bool {
+            true
+        }
+    }
+
+    fn pipeline_with(targets: Vec<MockTarget>) -> AuditPipeline {
+        let mut registry = AuditRegistry::new();
+        for target in targets {
+            registry.add_target(target.id.to_string(), Box::new(target));
+        }
+        AuditPipeline::new(Arc::new(Mutex::new(registry)))
+    }
+
+    fn entry() -> Arc<AuditEntry> {
+        Arc::new(AuditEntry::default())
+    }
+
+    // backlog#962: when every target rejects the event it is lost outright, so
+    // dispatch must return Err rather than swallowing the failures as Ok.
+    #[tokio::test]
+    async fn dispatch_returns_err_when_all_targets_fail() {
+        let pipeline = pipeline_with(vec![MockTarget::new("a:webhook", true), MockTarget::new("b:webhook", true)]);
+        let result = pipeline.dispatch(entry()).await;
+        assert!(matches!(result, Err(AuditError::Target(_))), "expected Err, got {result:?}");
+    }
+
+    // A partially-successful fan-out means the entry reached at least one sink,
+    // so dispatch reports success (degradation is logged, not propagated).
+    #[tokio::test]
+    async fn dispatch_returns_ok_on_partial_failure() {
+        let pipeline = pipeline_with(vec![MockTarget::new("ok:webhook", false), MockTarget::new("bad:webhook", true)]);
+        pipeline.dispatch(entry()).await.expect("partial success should return Ok");
+    }
+
+    #[tokio::test]
+    async fn dispatch_returns_ok_when_all_targets_succeed() {
+        let pipeline = pipeline_with(vec![MockTarget::new("a:webhook", false), MockTarget::new("b:webhook", false)]);
+        pipeline.dispatch(entry()).await.expect("all-success should return Ok");
+    }
+
+    // No configured targets is a benign no-op, not a failure.
+    #[tokio::test]
+    async fn dispatch_returns_ok_with_no_targets() {
+        let pipeline = pipeline_with(vec![]);
+        pipeline.dispatch(entry()).await.expect("no targets should return Ok");
+    }
+
+    // backlog#962: dispatch_batch must mirror dispatch and propagate a
+    // whole-batch loss instead of returning Ok.
+    #[tokio::test]
+    async fn dispatch_batch_returns_err_when_all_targets_fail() {
+        let pipeline = pipeline_with(vec![MockTarget::new("a:webhook", true)]);
+        let result = pipeline.dispatch_batch(vec![entry(), entry()]).await;
+        assert!(matches!(result, Err(AuditError::Target(_))), "expected Err, got {result:?}");
+    }
+
+    #[tokio::test]
+    async fn dispatch_batch_returns_ok_when_all_targets_succeed() {
+        let pipeline = pipeline_with(vec![MockTarget::new("a:webhook", false), MockTarget::new("b:webhook", false)]);
+        pipeline
+            .dispatch_batch(vec![entry(), entry()])
+            .await
+            .expect("all-success batch should return Ok");
+    }
+}

crates/audit/src/system.rs

@@ -89,14 +89,20 @@ impl AuditSystem {
         registry.create_audit_targets_from_config(config).await
     }
 
+    /// Stops any active replay workers and closes the currently installed
+    /// targets without touching the system state. Lock order is `registry`
+    /// then `stream_cancellers` to stay consistent with every other path that
+    /// holds both locks (see `runtime_status_snapshot`, backlog#961).
+    async fn shutdown_runtime_targets(&self) -> AuditResult<()> {
+        let mut registry = self.registry.lock().await;
+        let mut replay_workers = self.stream_cancellers.write().await;
+        self.runtime_facade()
+            .shutdown_runtime(&mut registry, &mut replay_workers)
+            .await
+    }
+
     async fn clear_runtime_targets(&self) -> AuditResult<()> {
-        {
-            let mut registry = self.registry.lock().await;
-            let mut replay_workers = self.stream_cancellers.write().await;
-            self.runtime_facade()
-                .shutdown_runtime(&mut registry, &mut replay_workers)
-                .await?;
-        }
+        self.shutdown_runtime_targets().await?;
 
         let mut state = self.state.write().await;
         *state = AuditSystemState::Stopped;
@@ -123,6 +129,16 @@ impl AuditSystem {
             "audit system state"
         );
 
+        // Stop-before-start (backlog#970): tear down the existing replay workers
+        // and close the currently installed targets *before* activating the new
+        // set. Activation spawns fresh replay workers for store-backed targets,
+        // so if the old workers were still running they would drain the same
+        // persistent queue concurrently with the new ones and re-deliver
+        // entries. Shutting the old runtime down first keeps at most one active
+        // worker per store across a reload. `replace_targets` performs a second
+        // (now no-op) shutdown before installing, which is idempotent.
+        self.shutdown_runtime_targets().await?;
+
         let activation = self.runtime_facade().activate_targets_with_replay(targets).await;
         self.runtime_facade().replace_targets(activation).await?;
 
@@ -139,21 +155,30 @@ impl AuditSystem {
     /// # Returns
     /// * `AuditResult<()>` - Result indicating success or failure
     pub async fn start(&self, config: Config) -> AuditResult<()> {
-        let state = self.state.write().await;
+        // Claim the `Starting` transition atomically while holding the write
+        // lock (backlog#978): the previous code released the lock after the
+        // check and re-acquired it later to set `Starting`, so two concurrent
+        // `start()` calls (or `start()` racing `reload`) could both pass the
+        // check and double-activate. Transitioning to `Starting` before
+        // dropping the guard makes a concurrent caller observe `Starting` and
+        // return early instead.
+        {
+            let mut state = self.state.write().await;
 
-        match *state {
-            AuditSystemState::Running => {
-                return Err(AuditError::AlreadyInitialized);
+            match *state {
+                AuditSystemState::Running => {
+                    return Err(AuditError::AlreadyInitialized);
+                }
+                AuditSystemState::Starting => {
+                    warn_audit_state("starting", Some("already_starting"));
+                    return Ok(());
+                }
+                _ => {}
             }
-            AuditSystemState::Starting => {
-                warn_audit_state("starting", Some("already_starting"));
-                return Ok(());
-            }
-            _ => {}
+
+            *state = AuditSystemState::Starting;
         }
 
-        drop(state);
-
         info!(
             event = EVENT_AUDIT_SYSTEM_STATE,
             component = LOG_COMPONENT_AUDIT,
@@ -173,11 +198,7 @@ impl AuditSystem {
 
         match self.create_targets_from_config(&config).await {
             Ok(targets) => {
-                {
-                    let mut state = self.state.write().await;
-                    *state = AuditSystemState::Starting;
-                }
-
+                // State is already `Starting` (claimed atomically above).
                 self.commit_runtime_targets(targets, AuditSystemState::Running).await?;
                 info_audit_state("running", None, None);
                 Ok(())
@@ -318,7 +339,13 @@ impl AuditSystem {
         match *state {
             AuditSystemState::Running => {}
             AuditSystemState::Paused => {
-                return Ok(());
+                // Do not silently return Ok while paused (backlog#978): the
+                // entry is neither delivered nor persisted, so reporting success
+                // would corrupt the audit trail. Surface an explicit `Paused`
+                // error and let the caller apply its policy (the global helper
+                // treats this as a deliberate skip; direct API callers can
+                // decide otherwise).
+                return Err(AuditError::Paused);
             }
             _ => {
                 return Err(AuditError::NotInitialized("Audit system is not running".to_string()));
@@ -548,6 +575,7 @@ fn warn_audit_state(state: &str, reason: Option<&str>) {
 #[cfg(test)]
 mod tests {
     use super::{AuditSystem, AuditSystemState};
+    use crate::{AuditEntry, AuditError};
     use async_trait::async_trait;
     use rustfs_targets::ReplayWorkerManager;
     use rustfs_targets::arn::TargetID;
@@ -705,4 +733,95 @@ mod tests {
             .await
             .expect("audit lock paths deadlocked (backlog#961 regression)");
     }
+
+    /// backlog#978: a paused system must not report success while silently
+    /// dropping the entry. `dispatch` should surface an explicit `Paused` error.
+    #[tokio::test]
+    async fn dispatch_while_paused_returns_error_not_ok() {
+        let system = AuditSystem::new();
+        {
+            let mut state = system.state.write().await;
+            *state = AuditSystemState::Paused;
+        }
+
+        let result = system.dispatch(Arc::new(AuditEntry::default())).await;
+        assert!(
+            matches!(result, Err(AuditError::Paused)),
+            "paused dispatch must return Err(Paused), got {result:?}"
+        );
+    }
+
+    /// backlog#978: `start()` now claims the `Starting` transition atomically
+    /// under the state lock, so racing `start()` calls cannot both pass the
+    /// check and double-activate. Hammer concurrent starts and assert the
+    /// workload completes (no deadlock/panic) and converges to a consistent
+    /// final state.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn concurrent_start_does_not_hang_or_double_activate() {
+        use std::time::Duration;
+
+        let system = AuditSystem::new();
+        let mut handles = Vec::new();
+        for _ in 0..8 {
+            let s = system.clone();
+            handles.push(tokio::spawn(async move {
+                // Empty config activates no targets, so a completed start settles
+                // the system back to `Stopped`.
+                let _ = s.start(rustfs_config::server_config::Config(HashMap::new())).await;
+            }));
+        }
+
+        let workload = async {
+            for handle in handles {
+                handle.await.expect("start task panicked");
+            }
+        };
+        tokio::time::timeout(Duration::from_secs(30), workload)
+            .await
+            .expect("concurrent start deadlocked (backlog#978 regression)");
+
+        assert_eq!(system.get_state().await, AuditSystemState::Stopped);
+    }
+
+    /// backlog#970: a reload/commit must tear down the previous replay workers
+    /// and close the old targets before activating the replacement set, so the
+    /// old and new workers never drain the same store concurrently. Seed an old
+    /// target plus a replay worker, commit a new target, and assert the old one
+    /// was closed and its worker stopped while the new one is installed.
+    #[tokio::test]
+    async fn commit_closes_old_targets_before_installing_new() {
+        let system = AuditSystem::new();
+
+        let old = TestTarget::new("old", "webhook");
+        let old_close = Arc::clone(&old.close_calls);
+        {
+            let mut registry = system.registry.lock().await;
+            registry.add_target("old:webhook".to_string(), Box::new(old));
+        }
+        {
+            let mut replay_workers = system.stream_cancellers.write().await;
+            let (cancel_tx, _cancel_rx) = mpsc::channel(1);
+            replay_workers.insert("old:webhook".to_string(), cancel_tx);
+        }
+        {
+            let mut state = system.state.write().await;
+            *state = AuditSystemState::Running;
+        }
+
+        let new = TestTarget::new("new", "webhook");
+        let new_close = Arc::clone(&new.close_calls);
+        system
+            .commit_runtime_targets(vec![Box::new(new)], AuditSystemState::Running)
+            .await
+            .expect("commit should succeed");
+
+        // Old target closed exactly once during the pre-install shutdown.
+        assert_eq!(old_close.load(Ordering::SeqCst), 1);
+        // New target installed and left open.
+        assert_eq!(new_close.load(Ordering::SeqCst), 0);
+        assert_eq!(system.list_targets().await, vec!["new:webhook".to_string()]);
+        // Old replay worker stopped; the store-less new target adds none.
+        assert_eq!(system.runtime_status_snapshot().await.replay_worker_count, 0);
+        assert_eq!(system.get_state().await, AuditSystemState::Running);
+    }
 }