diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 9b78fc846..79641d2cb 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -59,7 +59,7 @@ use std::{ path::PathBuf, sync::{ Arc, - atomic::{AtomicU32, Ordering}, + atomic::{AtomicBool, AtomicU32, Ordering}, }, time::Duration, }; @@ -216,6 +216,8 @@ where #[derive(Debug)] pub struct RemoteDisk { + /// Stable identity for this handle instance; replacement handles receive a new identity. + handle_id: Uuid, pub id: Mutex>, pub addr: String, endpoint: Endpoint, @@ -226,9 +228,20 @@ pub struct RemoteDisk { health: Arc, /// Cancellation token for monitoring tasks cancel_token: CancellationToken, + recovery_monitor_active: Arc, data_transport: Arc, } +struct RecoveryMonitorLease { + active: Arc, +} + +impl Drop for RecoveryMonitorLease { + fn drop(&mut self) { + self.active.store(false, Ordering::Release); + } +} + // ── Connection lifecycle (grpc-optimization P3) ── /// Whether to prewarm the internode control channel in the background at construction (default off). @@ -368,14 +381,15 @@ impl RemoteDisk { .await } - fn recovery_monitor_span(addr: &str, endpoint: &Endpoint) -> tracing::Span { + fn recovery_monitor_span(addr: &str, endpoint: &Endpoint, handle_id: Uuid) -> tracing::Span { tracing::info_span!( "recovery-monitor", component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REMOTE_DISK, kind = "remote_disk", endpoint = %endpoint, - addr = %addr + addr = %addr, + handle_id = %handle_id ) } @@ -411,6 +425,7 @@ impl RemoteDisk { rustfs_utils::get_env_bool(ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, DEFAULT_RUSTFS_DRIVE_ACTIVE_MONITORING); let disk = Self { + handle_id: Uuid::new_v4(), id: Mutex::new(None), addr, endpoint: ep.clone(), @@ -418,6 +433,7 @@ impl RemoteDisk { health_check: opt.health_check && env_health_check, health: Arc::new(DiskHealthTracker::new()), cancel_token: CancellationToken::new(), + recovery_monitor_active: Arc::new(AtomicBool::new(false)), data_transport, }; record_drive_runtime_state(ep, RuntimeDriveHealthState::Online); @@ -435,6 +451,11 @@ impl RemoteDisk { self.health.runtime_state() } + #[cfg(test)] + fn recovery_monitor_is_active(&self) -> bool { + self.recovery_monitor_active.load(Ordering::Acquire) + } + pub fn offline_duration_secs(&self) -> Option { self.health.offline_duration().map(|duration| duration.as_secs()) } @@ -573,13 +594,40 @@ impl RemoteDisk { return; } - let addr = self.addr.clone(); - let endpoint = self.endpoint.clone(); - let health = Arc::clone(&self.health); - let cancel_token = self.cancel_token.clone(); - let span = Self::recovery_monitor_span(&addr, &endpoint); + Self::schedule_recovery_monitor( + self.addr.clone(), + self.endpoint.clone(), + self.handle_id, + Arc::clone(&self.health), + self.cancel_token.clone(), + Arc::clone(&self.recovery_monitor_active), + ); + } + + fn schedule_recovery_monitor( + addr: String, + endpoint: Endpoint, + handle_id: Uuid, + health: Arc, + cancel_token: CancellationToken, + active: Arc, + ) { + if active + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return; + } + let span = Self::recovery_monitor_span(&addr, &endpoint, handle_id); super::spawn_background_monitor(span, async move { - Self::monitor_remote_disk_recovery(addr, endpoint, health, cancel_token).await; + let lease = RecoveryMonitorLease { + active: Arc::clone(&active), + }; + Self::monitor_remote_disk_recovery(addr.clone(), endpoint.clone(), Arc::clone(&health), cancel_token.clone()).await; + drop(lease); + if !cancel_token.is_cancelled() && health.runtime_state() != RuntimeDriveHealthState::Online { + Self::schedule_recovery_monitor(addr, endpoint, handle_id, health, cancel_token, active); + } }); } @@ -588,7 +636,7 @@ impl RemoteDisk { let (tx, rx) = tokio::sync::oneshot::channel(); let endpoint = self.endpoint.clone(); let addr = self.addr.clone(); - let span = Self::recovery_monitor_span(&addr, &endpoint); + let span = Self::recovery_monitor_span(&addr, &endpoint, self.handle_id); super::spawn_background_monitor(span, async move { warn!( event = EVENT_REMOTE_DISK_HEALTH, @@ -619,9 +667,11 @@ impl RemoteDisk { let cancel_token = self.cancel_token.clone(); let addr = self.addr.clone(); let endpoint = self.endpoint.clone(); + let handle_id = self.handle_id; + let recovery_monitor_active = Arc::clone(&self.recovery_monitor_active); tokio::spawn(async move { - Self::monitor_remote_disk_health(addr, endpoint, health, cancel_token).await; + Self::monitor_remote_disk_health(addr, endpoint, handle_id, health, cancel_token, recovery_monitor_active).await; }); } @@ -629,8 +679,10 @@ impl RemoteDisk { async fn monitor_remote_disk_health( addr: String, endpoint: Endpoint, + handle_id: Uuid, health: Arc, cancel_token: CancellationToken, + recovery_monitor_active: Arc, ) { let mut interval = time::interval(get_drive_active_check_interval()); @@ -655,11 +707,14 @@ impl RemoteDisk { let addr_clone = addr.clone(); let endpoint_clone = endpoint.clone(); let cancel_clone = cancel_token.clone(); - let span = Self::recovery_monitor_span(&addr_clone, &endpoint_clone); - - super::spawn_background_monitor(span, async move { - Self::monitor_remote_disk_recovery(addr_clone, endpoint_clone, health_clone, cancel_clone).await; - }); + Self::schedule_recovery_monitor( + addr_clone, + endpoint_clone, + handle_id, + health_clone, + cancel_clone, + Arc::clone(&recovery_monitor_active), + ); } loop { @@ -718,11 +773,14 @@ impl RemoteDisk { let addr_clone = addr.clone(); let endpoint_clone = endpoint.clone(); let cancel_clone = cancel_token.clone(); - let span = Self::recovery_monitor_span(&addr_clone, &endpoint_clone); - - super::spawn_background_monitor(span, async move { - Self::monitor_remote_disk_recovery(addr_clone, endpoint_clone, health_clone, cancel_clone).await; - }); + Self::schedule_recovery_monitor( + addr_clone, + endpoint_clone, + handle_id, + health_clone, + cancel_clone, + Arc::clone(&recovery_monitor_active), + ); } } } @@ -973,6 +1031,7 @@ impl RemoteDisk { subsystem = LOG_SUBSYSTEM_REMOTE_DISK, endpoint = %self.endpoint, addr = %self.addr, + handle_id = %self.handle_id, op, state = "faulty_short_circuit", "Remote disk operation short-circuited by faulty state" @@ -4382,6 +4441,45 @@ mod tests { accept_task.abort(); } + #[tokio::test] + async fn faulty_handle_runs_only_one_recovery_monitor() { + let endpoint = Endpoint { + url: url::Url::parse("http://remote-node:9000/data/rustfs0").expect("endpoint should parse"), + is_local: false, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }; + let disk = RemoteDisk::new( + &endpoint, + &DiskOption { + cleanup: false, + health_check: true, + }, + Arc::new(TcpHttpInternodeDataTransport), + ) + .await + .expect("remote disk should construct"); + if !disk.health_check { + return; + } + + disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline); + disk.spawn_recovery_monitor_if_needed(); + disk.spawn_recovery_monitor_if_needed(); + assert!(disk.recovery_monitor_is_active(), "only one recovery monitor should own the handle"); + + disk.cancel_token.cancel(); + tokio::time::timeout(Duration::from_secs(1), async { + while disk.recovery_monitor_is_active() { + tokio::task::yield_now().await; + } + }) + .await + .expect("cancelled recovery monitor should release its single-flight state"); + assert!(!disk.recovery_monitor_is_active()); + } + #[tokio::test] async fn test_copy_stream_with_buffer_copies_full_payload() { let payload = b"walk-dir-stream".repeat(1024); diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index deb40a5cb..0606b55e6 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -384,6 +384,17 @@ pub struct DiskHealthTracker { pub last_capacity_free: AtomicU64, /// Last successful capacity probe timestamp pub last_capacity_probe_unix_secs: AtomicI64, + /// Authoritative atomically published runtime/status pair. + state_snapshot: AtomicU64, + transition_lock: std::sync::Mutex<()>, +} + +fn pack_health_state(runtime_state: RuntimeDriveHealthState, status: u32) -> u64 { + (u64::from(runtime_state as u32) << 32) | u64::from(status) +} + +fn unpack_health_state(snapshot: u64) -> (RuntimeDriveHealthState, u32) { + (RuntimeDriveHealthState::from_u32((snapshot >> 32) as u32), snapshot as u32) } #[derive(Debug)] @@ -696,6 +707,8 @@ impl DiskHealthTracker { last_capacity_used: AtomicU64::new(0), last_capacity_free: AtomicU64::new(0), last_capacity_probe_unix_secs: AtomicI64::new(0), + state_snapshot: AtomicU64::new(pack_health_state(RuntimeDriveHealthState::Online, DISK_HEALTH_OK)), + transition_lock: std::sync::Mutex::new(()), } } @@ -732,38 +745,56 @@ impl DiskHealthTracker { /// Check if disk is faulty pub fn is_faulty(&self) -> bool { - self.status.load(Ordering::Acquire) == DISK_HEALTH_FAULTY + unpack_health_state(self.state_snapshot.load(Ordering::Acquire)).1 == DISK_HEALTH_FAULTY + } + + pub fn health_state_snapshot(&self) -> (RuntimeDriveHealthState, bool) { + let (runtime_state, status) = unpack_health_state(self.state_snapshot.load(Ordering::Acquire)); + (runtime_state, status == DISK_HEALTH_FAULTY) + } + + fn publish_state(&self, runtime_state: RuntimeDriveHealthState, status: u32) { + self.state_snapshot + .store(pack_health_state(runtime_state, status), Ordering::Release); + self.runtime_state.store(runtime_state as u32, Ordering::Release); + self.status.store(status, Ordering::Release); } /// Set disk as faulty pub fn set_faulty(&self) { - self.status.store(DISK_HEALTH_FAULTY, Ordering::Release); + let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + self.publish_state(RuntimeDriveHealthState::Offline, DISK_HEALTH_FAULTY); } /// Set disk as OK pub fn set_ok(&self) { - self.status.store(DISK_HEALTH_OK, Ordering::Release); + let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + self.publish_state(RuntimeDriveHealthState::Online, DISK_HEALTH_OK); } #[cfg(test)] pub fn force_runtime_state_for_test(&self, state: RuntimeDriveHealthState) { - self.runtime_state.store(state as u32, Ordering::Release); - match state { - RuntimeDriveHealthState::Offline => self.set_faulty(), - RuntimeDriveHealthState::Online | RuntimeDriveHealthState::Suspect | RuntimeDriveHealthState::Returning => { - self.set_ok(); - } - } + let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let status = if state == RuntimeDriveHealthState::Offline { + DISK_HEALTH_FAULTY + } else { + DISK_HEALTH_OK + }; + self.publish_state(state, status); } pub fn swap_ok_to_faulty(&self) -> bool { - self.status - .compare_exchange(DISK_HEALTH_OK, DISK_HEALTH_FAULTY, Ordering::AcqRel, Ordering::Relaxed) - .is_ok() + let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let (_, status) = unpack_health_state(self.state_snapshot.load(Ordering::Acquire)); + if status != DISK_HEALTH_OK { + return false; + } + self.publish_state(RuntimeDriveHealthState::Offline, DISK_HEALTH_FAULTY); + true } pub fn runtime_state(&self) -> RuntimeDriveHealthState { - RuntimeDriveHealthState::from_u32(self.runtime_state.load(Ordering::Acquire)) + unpack_health_state(self.state_snapshot.load(Ordering::Acquire)).0 } pub fn offline_duration(&self) -> Option { @@ -779,6 +810,7 @@ impl DiskHealthTracker { } pub fn mark_failure(&self, endpoint: &Endpoint, reason: &'static str) -> bool { + let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); let current = self.runtime_state(); let now = current_unix_secs(); let next = match current { @@ -807,23 +839,18 @@ impl DiskHealthTracker { }; let became_offline = next == RuntimeDriveHealthState::Offline && current != RuntimeDriveHealthState::Offline; - if next == RuntimeDriveHealthState::Offline { - self.status.store(DISK_HEALTH_FAULTY, Ordering::Release); - } else { - self.status.store(DISK_HEALTH_OK, Ordering::Release); - } self.transition_state(endpoint, current, next, reason); became_offline } pub fn mark_offline(&self, endpoint: &Endpoint, reason: &'static str) -> bool { + let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); let current = self.runtime_state(); if current == RuntimeDriveHealthState::Offline { return false; } self.consecutive_successes.store(0, Ordering::Release); - self.status.store(DISK_HEALTH_FAULTY, Ordering::Release); self.transition_state(endpoint, current, RuntimeDriveHealthState::Offline, reason); true } @@ -837,11 +864,10 @@ impl DiskHealthTracker { } fn reset_for_store_init_retry_at(&self, endpoint: &Endpoint, now: Duration) { + let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); let now_nanos = unix_nanos(now); let now_secs = unix_secs_i64(now); - self.status.store(DISK_HEALTH_OK, Ordering::Release); - self.runtime_state - .store(RuntimeDriveHealthState::Online as u32, Ordering::Release); + self.publish_state(RuntimeDriveHealthState::Online, DISK_HEALTH_OK); self.consecutive_failures.store(0, Ordering::Release); self.consecutive_successes.store(0, Ordering::Release); self.offline_since_unix_secs.store(0, Ordering::Release); @@ -853,6 +879,7 @@ impl DiskHealthTracker { } pub fn mark_recovery_success(&self, endpoint: &Endpoint, reason: &'static str) -> bool { + let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); let current = self.runtime_state(); let next = match current { RuntimeDriveHealthState::Online => RuntimeDriveHealthState::Online, @@ -873,7 +900,6 @@ impl DiskHealthTracker { let became_online = next == RuntimeDriveHealthState::Online; if became_online { - self.status.store(DISK_HEALTH_OK, Ordering::Release); self.consecutive_failures.store(0, Ordering::Release); self.consecutive_successes.store(0, Ordering::Release); } @@ -903,7 +929,13 @@ impl DiskHealthTracker { return; } - self.runtime_state.store(next as u32, Ordering::Release); + let current_status = unpack_health_state(self.state_snapshot.load(Ordering::Acquire)).1; + let status = match next { + RuntimeDriveHealthState::Offline => DISK_HEALTH_FAULTY, + RuntimeDriveHealthState::Returning => current_status, + RuntimeDriveHealthState::Online | RuntimeDriveHealthState::Suspect => DISK_HEALTH_OK, + }; + self.publish_state(next, status); self.last_transition_unix_secs .store(current_unix_secs() as i64, Ordering::Release); @@ -1189,7 +1221,7 @@ impl LocalDiskWrapper { return; } - if health.status.load(Ordering::Relaxed) != DISK_HEALTH_OK { + if health.is_faulty() { continue; } @@ -2900,6 +2932,35 @@ mod tests { }); } + #[test] + fn concurrent_failure_and_recovery_publish_one_health_snapshot() { + let endpoint = Endpoint::try_from("/tmp/concurrent-health-snapshot").expect("endpoint should parse"); + let health = Arc::new(DiskHealthTracker::new()); + let workers = (0..8) + .map(|_| { + let health = Arc::clone(&health); + let endpoint = endpoint.clone(); + std::thread::spawn(move || { + for _ in 0..32 { + health.mark_failure(&endpoint, "concurrent_test"); + health.mark_recovery_success(&endpoint, "concurrent_test"); + let (runtime, faulty) = health.health_state_snapshot(); + assert!(matches!( + (runtime, faulty), + (RuntimeDriveHealthState::Online, false) + | (RuntimeDriveHealthState::Suspect, false) + | (RuntimeDriveHealthState::Offline, true) + | (RuntimeDriveHealthState::Returning, true) + )); + } + }) + }) + .collect::>(); + for worker in workers { + worker.join().expect("health transition worker should not panic"); + } + } + #[test] fn operation_success_recovers_suspect_drive_without_faulting() { let endpoint = Endpoint::try_from("/tmp/runtime-state-suspect-success").expect("endpoint should parse");