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/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);
+    }
 }