rustfs_audit/
system.rs

1//  Copyright 2024 RustFS Team
2//
3//  Licensed under the Apache License, Version 2.0 (the "License");
4//  you may not use this file except in compliance with the License.
5//  You may obtain a copy of the License at
6//
7//      http://www.apache.org/licenses/LICENSE-2.0
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14
15use crate::{
16    AuditEntry, AuditError, AuditRegistry, AuditResult, observability,
17    pipeline::{AuditPipeline, AuditRuntimeFacade, AuditRuntimeView},
18};
19use rustfs_config::server_config::Config;
20use rustfs_targets::{ReplayWorkerManager, Target};
21use std::sync::Arc;
22use tokio::sync::{Mutex, RwLock};
23use tracing::{debug, error, info, warn};
24
25const LOG_COMPONENT_AUDIT: &str = "audit";
26const LOG_SUBSYSTEM_SYSTEM: &str = "system";
27const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state";
28const EVENT_AUDIT_CONFIG_RELOADED: &str = "audit_config_reloaded";
29
30#[derive(Debug, Clone, Default, PartialEq, Eq)]
31pub struct AuditTargetMetricSnapshot {
32    pub failed_messages: u64,
33    pub failed_store_length: u64,
34    pub queue_length: u64,
35    pub target_id: String,
36    pub total_messages: u64,
37}
38
39/// State of the audit system
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub enum AuditSystemState {
42    Stopped,
43    Starting,
44    Running,
45    Paused,
46    Stopping,
47}
48
49/// Main audit system that manages target lifecycle and audit log dispatch
50#[derive(Clone)]
51pub struct AuditSystem {
52    registry: Arc<Mutex<AuditRegistry>>,
53    state: Arc<RwLock<AuditSystemState>>,
54    config: Arc<RwLock<Option<Config>>>,
55    /// Cancellation senders for active audit stream tasks (target_id -> cancel tx)
56    stream_cancellers: Arc<RwLock<ReplayWorkerManager>>,
57}
58
59impl Default for AuditSystem {
60    fn default() -> Self {
61        Self::new()
62    }
63}
64
65impl AuditSystem {
66    fn pipeline(&self) -> AuditPipeline {
67        AuditPipeline::new(self.registry.clone())
68    }
69
70    fn runtime_view(&self) -> AuditRuntimeView {
71        AuditRuntimeView::new(self.registry.clone())
72    }
73
74    fn runtime_facade(&self) -> AuditRuntimeFacade {
75        AuditRuntimeFacade::new(self.registry.clone(), self.stream_cancellers.clone())
76    }
77
78    /// Creates a new audit system
79    pub fn new() -> Self {
80        Self {
81            registry: Arc::new(Mutex::new(AuditRegistry::new())),
82            state: Arc::new(RwLock::new(AuditSystemState::Stopped)),
83            config: Arc::new(RwLock::new(None)),
84            stream_cancellers: Arc::new(RwLock::new(ReplayWorkerManager::new())),
85        }
86    }
87
88    async fn create_targets_from_config(&self, config: &Config) -> AuditResult<Vec<Box<dyn Target<AuditEntry> + Send + Sync>>> {
89        let registry = self.registry.lock().await;
90        registry.create_audit_targets_from_config(config).await
91    }
92
93    /// Stops any active replay workers and closes the currently installed
94    /// targets without touching the system state. Lock order is `registry`
95    /// then `stream_cancellers` to stay consistent with every other path that
96    /// holds both locks (see `runtime_status_snapshot`, backlog#961).
97    async fn shutdown_runtime_targets(&self) -> AuditResult<()> {
98        let mut registry = self.registry.lock().await;
99        let mut replay_workers = self.stream_cancellers.write().await;
100        self.runtime_facade()
101            .shutdown_runtime(&mut registry, &mut replay_workers)
102            .await
103    }
104
105    async fn clear_runtime_targets(&self) -> AuditResult<()> {
106        self.shutdown_runtime_targets().await?;
107
108        let mut state = self.state.write().await;
109        *state = AuditSystemState::Stopped;
110        Ok(())
111    }
112
113    async fn commit_runtime_targets(
114        &self,
115        targets: Vec<Box<dyn Target<AuditEntry> + Send + Sync>>,
116        final_state: AuditSystemState,
117    ) -> AuditResult<()> {
118        if targets.is_empty() {
119            debug_audit_state("stopped", Some("no_enabled_targets"), None, 0);
120            self.clear_runtime_targets().await?;
121            return Ok(());
122        }
123
124        info!(
125            event = EVENT_AUDIT_SYSTEM_STATE,
126            component = LOG_COMPONENT_AUDIT,
127            subsystem = LOG_SUBSYSTEM_SYSTEM,
128            state = "targets_created",
129            target_count = targets.len(),
130            "audit system state"
131        );
132
133        // Stop-before-start (backlog#970): tear down the existing replay workers
134        // and close the currently installed targets *before* activating the new
135        // set. Activation spawns fresh replay workers for store-backed targets,
136        // so if the old workers were still running they would drain the same
137        // persistent queue concurrently with the new ones and re-deliver
138        // entries. Shutting the old runtime down first keeps at most one active
139        // worker per store across a reload. `replace_targets` performs a second
140        // (now no-op) shutdown before installing, which is idempotent.
141        self.shutdown_runtime_targets().await?;
142
143        let activation = self.runtime_facade().activate_targets_with_replay(targets).await;
144        self.runtime_facade().replace_targets(activation).await?;
145
146        let mut state = self.state.write().await;
147        *state = final_state;
148        Ok(())
149    }
150
151    /// Starts the audit system with the given configuration
152    ///
153    /// # Arguments
154    /// * `config` - The configuration to use for starting the audit system
155    ///
156    /// # Returns
157    /// * `AuditResult<()>` - Result indicating success or failure
158    pub async fn start(&self, config: Config) -> AuditResult<()> {
159        // Claim the `Starting` transition atomically while holding the write
160        // lock (backlog#978): the previous code released the lock after the
161        // check and re-acquired it later to set `Starting`, so two concurrent
162        // `start()` calls (or `start()` racing `reload`) could both pass the
163        // check and double-activate. Transitioning to `Starting` before
164        // dropping the guard makes a concurrent caller observe `Starting` and
165        // return early instead.
166        {
167            let mut state = self.state.write().await;
168
169            match *state {
170                AuditSystemState::Running => {
171                    return Err(AuditError::AlreadyInitialized);
172                }
173                AuditSystemState::Starting => {
174                    warn_audit_state("starting", Some("already_starting"));
175                    return Ok(());
176                }
177                _ => {}
178            }
179
180            *state = AuditSystemState::Starting;
181        }
182
183        info!(
184            event = EVENT_AUDIT_SYSTEM_STATE,
185            component = LOG_COMPONENT_AUDIT,
186            subsystem = LOG_SUBSYSTEM_SYSTEM,
187            state = "starting",
188            "audit system state"
189        );
190
191        // Record system start
192        observability::record_system_start();
193
194        // Store configuration
195        {
196            let mut config_guard = self.config.write().await;
197            *config_guard = Some(config.clone());
198        }
199
200        match self.create_targets_from_config(&config).await {
201            Ok(targets) => {
202                // State is already `Starting` (claimed atomically above).
203                self.commit_runtime_targets(targets, AuditSystemState::Running).await?;
204                info_audit_state("running", None, None);
205                Ok(())
206            }
207            Err(e) => {
208                error!(
209                    event = EVENT_AUDIT_SYSTEM_STATE,
210                    component = LOG_COMPONENT_AUDIT,
211                    subsystem = LOG_SUBSYSTEM_SYSTEM,
212                    state = "stopped",
213                    reason = "target_creation_failed",
214                    error = %e,
215                    "Failed to create audit targets"
216                );
217                let mut state = self.state.write().await;
218                *state = AuditSystemState::Stopped;
219                Err(e)
220            }
221        }
222    }
223
224    /// Pauses the audit system
225    ///
226    /// # Returns
227    /// * `AuditResult<()>` - Result indicating success or failure
228    pub async fn pause(&self) -> AuditResult<()> {
229        let mut state = self.state.write().await;
230
231        match *state {
232            AuditSystemState::Running => {
233                *state = AuditSystemState::Paused;
234                info_audit_state("paused", None, None);
235                Ok(())
236            }
237            AuditSystemState::Paused => {
238                warn_audit_state("paused", Some("already_paused"));
239                Ok(())
240            }
241            _ => Err(AuditError::Configuration("Cannot pause audit system in current state".to_string(), None)),
242        }
243    }
244
245    /// Resumes the audit system
246    ///
247    /// # Returns
248    /// * `AuditResult<()>` - Result indicating success or failure
249    pub async fn resume(&self) -> AuditResult<()> {
250        let mut state = self.state.write().await;
251
252        match *state {
253            AuditSystemState::Paused => {
254                *state = AuditSystemState::Running;
255                info_audit_state("running", Some("resumed"), None);
256                Ok(())
257            }
258            AuditSystemState::Running => {
259                warn_audit_state("running", Some("already_running"));
260                Ok(())
261            }
262            _ => Err(AuditError::Configuration("Cannot resume audit system in current state".to_string(), None)),
263        }
264    }
265
266    /// Stops the audit system and closes all targets
267    ///
268    /// # Returns
269    /// * `AuditResult<()>` - Result indicating success or failure
270    pub async fn close(&self) -> AuditResult<()> {
271        let mut state = self.state.write().await;
272
273        match *state {
274            AuditSystemState::Stopped => {
275                warn_audit_state("stopped", Some("already_stopped"));
276                return Ok(());
277            }
278            AuditSystemState::Stopping => {
279                warn_audit_state("stopping", Some("already_stopping"));
280                return Ok(());
281            }
282            _ => {}
283        }
284
285        *state = AuditSystemState::Stopping;
286        drop(state);
287
288        info!(
289            event = EVENT_AUDIT_SYSTEM_STATE,
290            component = LOG_COMPONENT_AUDIT,
291            subsystem = LOG_SUBSYSTEM_SYSTEM,
292            state = "stopping",
293            "audit system state"
294        );
295
296        // Stop all stream tasks first
297        if let Err(e) = self.clear_runtime_targets().await {
298            error!(
299                event = EVENT_AUDIT_SYSTEM_STATE,
300                component = LOG_COMPONENT_AUDIT,
301                subsystem = LOG_SUBSYSTEM_SYSTEM,
302                state = "stopping",
303                reason = "target_shutdown_failed",
304                error = %e,
305                "Failed to close some audit targets"
306            );
307        }
308
309        // Clear configuration
310        let mut config_guard = self.config.write().await;
311        *config_guard = None;
312
313        info_audit_state("stopped", None, None);
314        Ok(())
315    }
316
317    /// Gets the current state of the audit system
318    pub async fn get_state(&self) -> AuditSystemState {
319        self.state.read().await.clone()
320    }
321
322    /// Checks if the audit system is running
323    ///
324    /// # Returns
325    /// * `bool` - True if running, false otherwise
326    pub async fn is_running(&self) -> bool {
327        matches!(*self.state.read().await, AuditSystemState::Running)
328    }
329
330    /// Dispatches an audit log entry to all active targets
331    ///
332    /// # Arguments
333    /// * `entry` - The audit log entry to dispatch
334    ///
335    /// # Returns
336    /// * `AuditResult<()>` - Result indicating success or failure
337    pub async fn dispatch(&self, entry: Arc<AuditEntry>) -> AuditResult<()> {
338        let state = self.state.read().await;
339
340        match *state {
341            AuditSystemState::Running => {}
342            AuditSystemState::Paused => {
343                // Do not silently return Ok while paused (backlog#978): the
344                // entry is neither delivered nor persisted, so reporting success
345                // would corrupt the audit trail. Surface an explicit `Paused`
346                // error and let the caller apply its policy (the global helper
347                // treats this as a deliberate skip; direct API callers can
348                // decide otherwise).
349                return Err(AuditError::Paused);
350            }
351            _ => {
352                return Err(AuditError::NotInitialized("Audit system is not running".to_string()));
353            }
354        }
355        drop(state);
356        self.pipeline().dispatch(entry).await
357    }
358
359    /// Dispatches a batch of audit log entries to all active targets
360    ///
361    /// # Arguments
362    /// * `entries` - A vector of audit log entries to dispatch
363    ///
364    /// # Returns
365    /// * `AuditResult<()>` - Result indicating success or failure
366    pub async fn dispatch_batch(&self, entries: Vec<Arc<AuditEntry>>) -> AuditResult<()> {
367        let state = self.state.read().await;
368        if *state != AuditSystemState::Running {
369            return Err(AuditError::NotInitialized("Audit system is not running".to_string()));
370        }
371        drop(state);
372        self.pipeline().dispatch_batch(entries).await
373    }
374
375    /// Enables a specific target
376    ///
377    /// # Arguments
378    /// * `target_id` - The ID of the target to enable, TargetID to string
379    ///
380    /// # Returns
381    /// * `AuditResult<()>` - Result indicating success or failure
382    pub async fn enable_target(&self, target_id: &str) -> AuditResult<()> {
383        self.runtime_view().enable_target(target_id).await
384    }
385
386    /// Disables a specific target
387    ///
388    /// # Arguments
389    /// * `target_id` - The ID of the target to disable, TargetID to string
390    ///
391    /// # Returns
392    /// * `AuditResult<()>` - Result indicating success or failure
393    pub async fn disable_target(&self, target_id: &str) -> AuditResult<()> {
394        self.runtime_view().disable_target(target_id).await
395    }
396
397    /// Removes a target from the system
398    ///
399    /// # Arguments
400    /// * `target_id` - The ID of the target to remove, TargetID to string
401    ///
402    /// # Returns
403    /// * `AuditResult<()>` - Result indicating success or failure
404    pub async fn remove_target(&self, target_id: &str) -> AuditResult<()> {
405        self.runtime_view().remove_target(target_id).await
406    }
407
408    /// Updates or inserts a target
409    ///
410    /// # Arguments
411    /// * `target_id` - The ID of the target to upsert, TargetID to string
412    /// * `target` - The target instance to insert or update
413    ///
414    /// # Returns
415    /// * `AuditResult<()>` - Result indicating success or failure
416    pub async fn upsert_target(&self, target_id: String, target: Box<dyn Target<AuditEntry> + Send + Sync>) -> AuditResult<()> {
417        self.runtime_view().upsert_target(target_id, target).await
418    }
419
420    /// Lists all targets
421    ///
422    /// # Returns
423    /// * `Vec<String>` - List of target IDs
424    pub async fn list_targets(&self) -> Vec<String> {
425        self.runtime_view().list_targets().await
426    }
427
428    /// Returns cloned target values for read-only runtime inspection.
429    pub async fn get_target_values(&self) -> Vec<rustfs_targets::SharedTarget<AuditEntry>> {
430        self.runtime_view().get_target_values().await
431    }
432
433    /// Returns per-target delivery metrics for Prometheus collection.
434    pub async fn snapshot_target_metrics(&self) -> Vec<AuditTargetMetricSnapshot> {
435        self.pipeline().snapshot_target_metrics().await
436    }
437
438    pub async fn snapshot_target_health(&self) -> Vec<rustfs_targets::RuntimeTargetHealthSnapshot> {
439        self.pipeline().snapshot_target_health().await
440    }
441
442    pub async fn runtime_status_snapshot(&self) -> rustfs_targets::RuntimeStatusSnapshot {
443        // Lock order must match every other path that holds both locks
444        // (`clear_runtime_targets`, `AuditRuntimeFacade::replace_targets`):
445        // acquire `registry` first, then `stream_cancellers`. Reversing the
446        // order here would create an ABBA deadlock with those paths.
447        let registry = self.registry.lock().await;
448        let replay_workers = self.stream_cancellers.read().await;
449        registry.runtime_manager().status_snapshot(&replay_workers)
450    }
451
452    /// Gets information about a specific target
453    ///
454    /// # Arguments
455    /// * `target_id` - The ID of the target to retrieve, TargetID to string
456    ///
457    /// # Returns
458    /// * `Option<String>` - Target ID if found
459    pub async fn get_target(&self, target_id: &str) -> Option<String> {
460        self.runtime_view().get_target(target_id).await
461    }
462
463    /// Reloads configuration and updates targets
464    ///
465    /// # Arguments
466    /// * `new_config` - The new configuration to load
467    ///
468    /// # Returns
469    /// * `AuditResult<()>` - Result indicating success or failure
470    pub async fn reload_config(&self, new_config: Config) -> AuditResult<()> {
471        info!(
472            event = EVENT_AUDIT_CONFIG_RELOADED,
473            component = LOG_COMPONENT_AUDIT,
474            subsystem = LOG_SUBSYSTEM_SYSTEM,
475            state = "reloading",
476            "audit config reload"
477        );
478
479        observability::record_config_reload();
480
481        // Store new configuration
482        {
483            let mut config_guard = self.config.write().await;
484            *config_guard = Some(new_config.clone());
485        }
486
487        let final_state = match self.get_state().await {
488            AuditSystemState::Paused => AuditSystemState::Paused,
489            _ => AuditSystemState::Running,
490        };
491
492        match self.create_targets_from_config(&new_config).await {
493            Ok(targets) => {
494                self.commit_runtime_targets(targets, final_state).await?;
495                info!(
496                    event = EVENT_AUDIT_CONFIG_RELOADED,
497                    component = LOG_COMPONENT_AUDIT,
498                    subsystem = LOG_SUBSYSTEM_SYSTEM,
499                    state = "reloaded",
500                    "audit config reload"
501                );
502                Ok(())
503            }
504            Err(e) => {
505                error!(
506                    event = EVENT_AUDIT_CONFIG_RELOADED,
507                    component = LOG_COMPONENT_AUDIT,
508                    subsystem = LOG_SUBSYSTEM_SYSTEM,
509                    state = "reload_failed",
510                    error = %e,
511                    "Failed to reload audit configuration"
512                );
513                Err(e)
514            }
515        }
516    }
517
518    /// Gets current audit system metrics
519    ///
520    /// # Returns
521    /// * `AuditMetricsReport` - Current metrics report
522    pub async fn get_metrics(&self) -> observability::AuditMetricsReport {
523        observability::get_metrics_report().await
524    }
525
526    /// Validates system performance against requirements
527    ///
528    /// # Returns
529    /// * `PerformanceValidation` - Performance validation results
530    pub async fn validate_performance(&self) -> observability::PerformanceValidation {
531        observability::validate_performance().await
532    }
533
534    /// Resets all metrics to initial state
535    pub async fn reset_metrics(&self) {
536        observability::reset_metrics().await;
537    }
538}
539
540fn info_audit_state(state: &str, reason: Option<&str>, target_count: Option<usize>) {
541    info!(
542        event = EVENT_AUDIT_SYSTEM_STATE,
543        component = LOG_COMPONENT_AUDIT,
544        subsystem = LOG_SUBSYSTEM_SYSTEM,
545        state,
546        reason = reason.unwrap_or_default(),
547        target_count = target_count.unwrap_or_default(),
548        "audit system state"
549    );
550}
551
552fn debug_audit_state(state: &str, reason: Option<&str>, error: Option<&str>, target_count: usize) {
553    debug!(
554        event = EVENT_AUDIT_SYSTEM_STATE,
555        component = LOG_COMPONENT_AUDIT,
556        subsystem = LOG_SUBSYSTEM_SYSTEM,
557        state,
558        reason = reason.unwrap_or_default(),
559        error = error.unwrap_or_default(),
560        target_count,
561        "audit system state"
562    );
563}
564
565fn warn_audit_state(state: &str, reason: Option<&str>) {
566    warn!(
567        event = EVENT_AUDIT_SYSTEM_STATE,
568        component = LOG_COMPONENT_AUDIT,
569        subsystem = LOG_SUBSYSTEM_SYSTEM,
570        state,
571        reason = reason.unwrap_or_default(),
572        "audit system state"
573    );
574}
575
576#[cfg(test)]
577mod tests {
578    use super::{AuditSystem, AuditSystemState};
579    use crate::{AuditEntry, AuditError};
580    use rustfs_targets::ReplayWorkerManager;
581    use rustfs_targets::testkit::MockTarget;
582    use std::collections::HashMap;
583    use std::sync::Arc;
584    use tokio::sync::mpsc;
585
586    #[tokio::test]
587    async fn reload_with_empty_config_stops_existing_runtime() {
588        let system = AuditSystem::new();
589        let target = MockTarget::new("primary", "webhook");
590        let observer = target.clone();
591
592        {
593            let mut registry = system.registry.lock().await;
594            registry.add_target("primary:webhook".to_string(), Box::new(target));
595        }
596        {
597            let mut state = system.state.write().await;
598            *state = AuditSystemState::Running;
599        }
600        {
601            let mut replay_workers = system.stream_cancellers.write().await;
602            let (cancel_tx, _cancel_rx) = mpsc::channel(1);
603            replay_workers.insert("primary:webhook".to_string(), cancel_tx);
604            assert_eq!(replay_workers.len(), 1);
605        }
606
607        system
608            .reload_config(rustfs_config::server_config::Config(HashMap::new()))
609            .await
610            .expect("reload with empty config should succeed");
611
612        assert_eq!(system.get_state().await, AuditSystemState::Stopped);
613        assert!(system.list_targets().await.is_empty());
614        assert_eq!(system.runtime_status_snapshot().await, ReplayWorkerManager::new().snapshot(0));
615        assert_eq!(observer.close_call_count(), 1);
616        assert_eq!(*system.config.read().await, Some(rustfs_config::server_config::Config(HashMap::new())));
617    }
618
619    /// Regression guard for backlog#961: `runtime_status_snapshot` and
620    /// `clear_runtime_targets` both hold `registry` and `stream_cancellers`.
621    /// They previously acquired the two locks in opposite orders (ABBA),
622    /// which could deadlock the whole audit control plane under concurrency.
623    /// Hammer both paths from multiple worker threads and assert the workload
624    /// completes within a timeout instead of hanging.
625    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
626    async fn concurrent_status_and_clear_do_not_deadlock() {
627        use std::time::Duration;
628
629        const ITERATIONS: usize = 2_000;
630        const TASKS_PER_PATH: usize = 4;
631
632        let system = AuditSystem::new();
633
634        // Seed a target + replay worker so both critical sections touch real state.
635        {
636            let mut registry = system.registry.lock().await;
637            registry.add_target("primary:webhook".to_string(), Box::new(MockTarget::new("primary", "webhook")));
638        }
639        {
640            let mut replay_workers = system.stream_cancellers.write().await;
641            let (cancel_tx, _cancel_rx) = mpsc::channel(1);
642            replay_workers.insert("primary:webhook".to_string(), cancel_tx);
643        }
644
645        let mut handles = Vec::new();
646
647        for _ in 0..TASKS_PER_PATH {
648            let status_system = system.clone();
649            handles.push(tokio::spawn(async move {
650                for _ in 0..ITERATIONS {
651                    // registry -> stream_cancellers (read)
652                    let _ = status_system.runtime_status_snapshot().await;
653                }
654            }));
655
656            let clear_system = system.clone();
657            handles.push(tokio::spawn(async move {
658                for _ in 0..ITERATIONS {
659                    // registry -> stream_cancellers (write)
660                    clear_system
661                        .clear_runtime_targets()
662                        .await
663                        .expect("clear_runtime_targets should succeed");
664                }
665            }));
666        }
667
668        let workload = async {
669            for handle in handles {
670                handle.await.expect("worker task panicked");
671            }
672        };
673
674        tokio::time::timeout(Duration::from_secs(30), workload)
675            .await
676            .expect("audit lock paths deadlocked (backlog#961 regression)");
677    }
678
679    /// backlog#978: a paused system must not report success while silently
680    /// dropping the entry. `dispatch` should surface an explicit `Paused` error.
681    #[tokio::test]
682    async fn dispatch_while_paused_returns_error_not_ok() {
683        let system = AuditSystem::new();
684        {
685            let mut state = system.state.write().await;
686            *state = AuditSystemState::Paused;
687        }
688
689        let result = system.dispatch(Arc::new(AuditEntry::default())).await;
690        assert!(
691            matches!(result, Err(AuditError::Paused)),
692            "paused dispatch must return Err(Paused), got {result:?}"
693        );
694    }
695
696    /// backlog#978: `start()` now claims the `Starting` transition atomically
697    /// under the state lock, so racing `start()` calls cannot both pass the
698    /// check and double-activate. Hammer concurrent starts and assert the
699    /// workload completes (no deadlock/panic) and converges to a consistent
700    /// final state.
701    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
702    async fn concurrent_start_does_not_hang_or_double_activate() {
703        use std::time::Duration;
704
705        let system = AuditSystem::new();
706        let mut handles = Vec::new();
707        for _ in 0..8 {
708            let s = system.clone();
709            handles.push(tokio::spawn(async move {
710                // Empty config activates no targets, so a completed start settles
711                // the system back to `Stopped`.
712                let _ = s.start(rustfs_config::server_config::Config(HashMap::new())).await;
713            }));
714        }
715
716        let workload = async {
717            for handle in handles {
718                handle.await.expect("start task panicked");
719            }
720        };
721        tokio::time::timeout(Duration::from_secs(30), workload)
722            .await
723            .expect("concurrent start deadlocked (backlog#978 regression)");
724
725        assert_eq!(system.get_state().await, AuditSystemState::Stopped);
726    }
727
728    /// backlog#970: a reload/commit must tear down the previous replay workers
729    /// and close the old targets before activating the replacement set, so the
730    /// old and new workers never drain the same store concurrently. Seed an old
731    /// target plus a replay worker, commit a new target, and assert the old one
732    /// was closed and its worker stopped while the new one is installed.
733    #[tokio::test]
734    async fn commit_closes_old_targets_before_installing_new() {
735        let system = AuditSystem::new();
736
737        let old = MockTarget::new("old", "webhook");
738        let old_observer = old.clone();
739        {
740            let mut registry = system.registry.lock().await;
741            registry.add_target("old:webhook".to_string(), Box::new(old));
742        }
743        {
744            let mut replay_workers = system.stream_cancellers.write().await;
745            let (cancel_tx, _cancel_rx) = mpsc::channel(1);
746            replay_workers.insert("old:webhook".to_string(), cancel_tx);
747        }
748        {
749            let mut state = system.state.write().await;
750            *state = AuditSystemState::Running;
751        }
752
753        let new = MockTarget::new("new", "webhook");
754        let new_observer = new.clone();
755        system
756            .commit_runtime_targets(vec![Box::new(new)], AuditSystemState::Running)
757            .await
758            .expect("commit should succeed");
759
760        // Old target closed exactly once during the pre-install shutdown.
761        assert_eq!(old_observer.close_call_count(), 1);
762        // New target installed and left open.
763        assert_eq!(new_observer.close_call_count(), 0);
764        assert_eq!(system.list_targets().await, vec!["new:webhook".to_string()]);
765        // Old replay worker stopped; the store-less new target adds none.
766        assert_eq!(system.runtime_status_snapshot().await.replay_worker_count, 0);
767        assert_eq!(system.get_state().await, AuditSystemState::Running);
768    }
769}
rustfs_audit/system.rs

rustfs_audit/
system.rs
