mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-27 08:38:58 +00:00
fix(audit): resolve ABBA lock-order deadlock between registry and stream_cancellers (#4421)
* fix(audit): resolve AB-BA lock-order deadlock between registry and stream_cancellers (backlog#961) `AuditSystem::runtime_status_snapshot` acquired `stream_cancellers.read()` before `registry.lock()`, while every other path that holds both locks (`clear_runtime_targets`, `AuditRuntimeFacade::replace_targets`) acquires `registry` first. Under concurrency (e.g. admin status polling racing a config reload) this AB-BA ordering could deadlock the whole audit control plane with no self-recovery. Reorder `runtime_status_snapshot` to acquire `registry` before `stream_cancellers`, unifying the global lock order to `registry -> stream_cancellers` across all double-lock paths, and add a multi-threaded regression test that hammers both critical sections with a timeout guard to fail fast if the ordering regresses. Refs: https://github.com/rustfs/backlog/issues/961 * docs(audit): reword AB-BA to ABBA to satisfy typos check (backlog#961)
This commit is contained in:
@@ -412,8 +412,12 @@ impl AuditSystem {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub async fn runtime_status_snapshot(&self) -> rustfs_targets::RuntimeStatusSnapshot {
|
pub async fn runtime_status_snapshot(&self) -> rustfs_targets::RuntimeStatusSnapshot {
|
||||||
let replay_workers = self.stream_cancellers.read().await;
|
// Lock order must match every other path that holds both locks
|
||||||
|
// (`clear_runtime_targets`, `AuditRuntimeFacade::replace_targets`):
|
||||||
|
// acquire `registry` first, then `stream_cancellers`. Reversing the
|
||||||
|
// order here would create an ABBA deadlock with those paths.
|
||||||
let registry = self.registry.lock().await;
|
let registry = self.registry.lock().await;
|
||||||
|
let replay_workers = self.stream_cancellers.read().await;
|
||||||
registry.runtime_manager().status_snapshot(&replay_workers)
|
registry.runtime_manager().status_snapshot(&replay_workers)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -641,4 +645,64 @@ mod tests {
|
|||||||
assert_eq!(close_calls.load(Ordering::SeqCst), 1);
|
assert_eq!(close_calls.load(Ordering::SeqCst), 1);
|
||||||
assert_eq!(*system.config.read().await, Some(rustfs_config::server_config::Config(HashMap::new())));
|
assert_eq!(*system.config.read().await, Some(rustfs_config::server_config::Config(HashMap::new())));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Regression guard for backlog#961: `runtime_status_snapshot` and
|
||||||
|
/// `clear_runtime_targets` both hold `registry` and `stream_cancellers`.
|
||||||
|
/// They previously acquired the two locks in opposite orders (ABBA),
|
||||||
|
/// which could deadlock the whole audit control plane under concurrency.
|
||||||
|
/// Hammer both paths from multiple worker threads and assert the workload
|
||||||
|
/// completes within a timeout instead of hanging.
|
||||||
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
||||||
|
async fn concurrent_status_and_clear_do_not_deadlock() {
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
const ITERATIONS: usize = 2_000;
|
||||||
|
const TASKS_PER_PATH: usize = 4;
|
||||||
|
|
||||||
|
let system = AuditSystem::new();
|
||||||
|
|
||||||
|
// Seed a target + replay worker so both critical sections touch real state.
|
||||||
|
{
|
||||||
|
let mut registry = system.registry.lock().await;
|
||||||
|
registry.add_target("primary:webhook".to_string(), Box::new(TestTarget::new("primary", "webhook")));
|
||||||
|
}
|
||||||
|
{
|
||||||
|
let mut replay_workers = system.stream_cancellers.write().await;
|
||||||
|
let (cancel_tx, _cancel_rx) = mpsc::channel(1);
|
||||||
|
replay_workers.insert("primary:webhook".to_string(), cancel_tx);
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut handles = Vec::new();
|
||||||
|
|
||||||
|
for _ in 0..TASKS_PER_PATH {
|
||||||
|
let status_system = system.clone();
|
||||||
|
handles.push(tokio::spawn(async move {
|
||||||
|
for _ in 0..ITERATIONS {
|
||||||
|
// registry -> stream_cancellers (read)
|
||||||
|
let _ = status_system.runtime_status_snapshot().await;
|
||||||
|
}
|
||||||
|
}));
|
||||||
|
|
||||||
|
let clear_system = system.clone();
|
||||||
|
handles.push(tokio::spawn(async move {
|
||||||
|
for _ in 0..ITERATIONS {
|
||||||
|
// registry -> stream_cancellers (write)
|
||||||
|
clear_system
|
||||||
|
.clear_runtime_targets()
|
||||||
|
.await
|
||||||
|
.expect("clear_runtime_targets should succeed");
|
||||||
|
}
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
|
||||||
|
let workload = async {
|
||||||
|
for handle in handles {
|
||||||
|
handle.await.expect("worker task panicked");
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
tokio::time::timeout(Duration::from_secs(30), workload)
|
||||||
|
.await
|
||||||
|
.expect("audit lock paths deadlocked (backlog#961 regression)");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user