fix(ecstore): single-flight remote disk recovery

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<Option<Uuid>>,
     pub addr: String,
     endpoint: Endpoint,
@@ -226,9 +228,20 @@ pub struct RemoteDisk {
     health: Arc<DiskHealthTracker>,
     /// Cancellation token for monitoring tasks
     cancel_token: CancellationToken,
+    recovery_monitor_active: Arc<AtomicBool>,
     data_transport: Arc<dyn InternodeDataTransport>,
 }
 
+struct RecoveryMonitorLease {
+    active: Arc<AtomicBool>,
+}
+
+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<u64> {
         self.health.offline_duration().map(|duration| duration.as_secs())
     }
@@ -573,13 +594,40 @@ impl RemoteDisk {
             return;
         }
 
-        let addr = self.addr.clone();
+        Self::schedule_recovery_monitor(
-        let endpoint = self.endpoint.clone();
+            self.addr.clone(),
-        let health = Arc::clone(&self.health);
+            self.endpoint.clone(),
-        let cancel_token = self.cancel_token.clone();
+            self.handle_id,
-        let span = Self::recovery_monitor_span(&addr, &endpoint);
+            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<DiskHealthTracker>,
+        cancel_token: CancellationToken,
+        active: Arc<AtomicBool>,
+    ) {
+        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<DiskHealthTracker>,
         cancel_token: CancellationToken,
+        recovery_monitor_active: Arc<AtomicBool>,
     ) {
         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);
+            Self::schedule_recovery_monitor(
-
+                addr_clone,
-            super::spawn_background_monitor(span, async move {
+                endpoint_clone,
-                Self::monitor_remote_disk_recovery(addr_clone, endpoint_clone, health_clone, cancel_clone).await;
+                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);
+                        Self::schedule_recovery_monitor(
-
+                            addr_clone,
-                        super::spawn_background_monitor(span, async move {
+                            endpoint_clone,
-                            Self::monitor_remote_disk_recovery(addr_clone, endpoint_clone, health_clone, cancel_clone).await;
+                            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);

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);
+        let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
-        match state {
+        let status = if state == RuntimeDriveHealthState::Offline {
-            RuntimeDriveHealthState::Offline => self.set_faulty(),
+            DISK_HEALTH_FAULTY
-            RuntimeDriveHealthState::Online | RuntimeDriveHealthState::Suspect | RuntimeDriveHealthState::Returning => {
+        } else {
-                self.set_ok();
+            DISK_HEALTH_OK
-            }
+        };
-        }
+        self.publish_state(state, status);
     }
 
     pub fn swap_ok_to_faulty(&self) -> bool {
-        self.status
+        let _guard = self.transition_lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
-            .compare_exchange(DISK_HEALTH_OK, DISK_HEALTH_FAULTY, Ordering::AcqRel, Ordering::Relaxed)
+        let (_, status) = unpack_health_state(self.state_snapshot.load(Ordering::Acquire));
-            .is_ok()
+        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<Duration> {
@@ -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.publish_state(RuntimeDriveHealthState::Online, DISK_HEALTH_OK);
-        self.runtime_state
-            .store(RuntimeDriveHealthState::Online as u32, Ordering::Release);
         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::<Vec<_>>();
+        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");