From e44bece00df347f074a1ff31be0774fcb7c3ec9e Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 9 Jul 2026 01:05:21 +0800 Subject: [PATCH] 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 --- crates/audit/src/error.rs | 3 + crates/audit/src/global.rs | 37 +++++--- crates/audit/src/pipeline.rs | 154 +++++++++++++++++++++++++++++++- crates/audit/src/system.rs | 167 ++++++++++++++++++++++++++++++----- 4 files changed, 321 insertions(+), 40 deletions(-) diff --git a/crates/audit/src/error.rs b/crates/audit/src/error.rs index 3734437ee..98f9ba1bb 100644 --- a/crates/audit/src/error.rs +++ b/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), diff --git a/crates/audit/src/global.rs b/crates/audit/src/global.rs index eedd8805a..dd61e5d36 100644 --- a/crates/audit/src/global.rs +++ b/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) -> 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) -> 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), } } diff --git a/crates/audit/src/pipeline.rs b/crates/audit/src/pipeline.rs index df5f04fdd..9325e0cc4 100644 --- a/crates/audit/src/pipeline.rs +++ b/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) -> 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 Target for MockTarget + where + E: rustfs_targets::PluginEvent, + { + fn id(&self) -> TargetID { + self.id.clone() + } + + async fn is_active(&self) -> Result { + Ok(true) + } + + async fn save(&self, _event: Arc>) -> 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, _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()) + } + + fn is_enabled(&self) -> bool { + true + } + } + + fn pipeline_with(targets: Vec) -> 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 { + 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"); + } +} diff --git a/crates/audit/src/system.rs b/crates/audit/src/system.rs index 35ed891a1..ead1f5157 100644 --- a/crates/audit/src/system.rs +++ b/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); + } }