fix(ecstore): single-flight remote disk recovery (#6096)

* fix(ecstore): single-flight remote disk recovery

* test(ecstore): cover remote recovery review cases

* test(ecstore): exercise recovery through disk slot

* test(ecstore): match format reads exactly

* test(ecstore): cover recovery teardown races

* style(ecstore): format recovery race tests

* fix(ecstore): remove unused health snapshot helper

* fix(ecstore): group recovery monitor test state

* fix(protos): preserve production source in compatibility checks

Cargo.lock

@@ -9465,6 +9465,7 @@ dependencies = [
  "tokio-stream",
  "tokio-util",
  "tonic",
+ "tonic-prost",
  "tower",
  "tracing",
  "tracing-core",

crates/ecstore/Cargo.toml

@@ -273,6 +273,7 @@ proptest = "1"
 rcgen.workspace = true
 insta = { workspace = true, features = ["yaml", "json"] }
 rustfs-crypto = { workspace = true }
+tonic-prost = { workspace = true }
 
 [build-dependencies]
 shadow-rs = { workspace = true, default-features = false, features = ["build", "metadata"] }

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -61,7 +61,7 @@ use std::{
     pin::Pin,
     sync::{
         Arc,
-        atomic::{AtomicU32, Ordering},
+        atomic::{AtomicBool, AtomicU32, Ordering},
     },
     task::{Context, Poll},
     time::Duration,
@@ -630,6 +630,8 @@ impl rustfs_rio::ChunkReader for RetryingRemoteChunkReader {
 
 #[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,
@@ -640,9 +642,38 @@ pub struct RemoteDisk {
     health: Arc<DiskHealthTracker>,
     /// Cancellation token for monitoring tasks
     cancel_token: CancellationToken,
+    recovery_monitor_active: Arc<AtomicBool>,
+    #[cfg(test)]
+    recovery_monitor_start_count: Arc<AtomicU32>,
+    #[cfg(test)]
+    recovery_monitor_teardown_hook: Arc<tokio::sync::Mutex<Option<Arc<RecoveryMonitorTeardownHook>>>>,
     data_transport: Arc<dyn InternodeDataTransport>,
 }
 
+struct RecoveryMonitorLease {
+    active: Arc<AtomicBool>,
+}
+
+impl Drop for RecoveryMonitorLease {
+    fn drop(&mut self) {
+        self.active.store(false, Ordering::Release);
+    }
+}
+
+#[cfg(test)]
+#[derive(Debug, Default)]
+struct RecoveryMonitorTeardownHook {
+    arrived: tokio::sync::Notify,
+    release: tokio::sync::Notify,
+}
+
+#[cfg(test)]
+#[derive(Clone)]
+struct RecoveryMonitorTestState {
+    start_count: Arc<AtomicU32>,
+    teardown_hook: Arc<tokio::sync::Mutex<Option<Arc<RecoveryMonitorTeardownHook>>>>,
+}
+
 // ── Connection lifecycle (grpc-optimization P3) ──
 
 /// Whether to prewarm the internode control channel in the background at construction (default off).
@@ -782,14 +813,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
         )
     }
 
@@ -825,6 +857,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(),
@@ -832,6 +865,11 @@ 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)),
+            #[cfg(test)]
+            recovery_monitor_start_count: Arc::new(AtomicU32::new(0)),
+            #[cfg(test)]
+            recovery_monitor_teardown_hook: Arc::new(tokio::sync::Mutex::new(None)),
             data_transport,
         };
         record_drive_runtime_state(ep, RuntimeDriveHealthState::Online);
@@ -849,6 +887,16 @@ impl RemoteDisk {
         self.health.runtime_state()
     }
 
+    #[cfg(test)]
+    fn recovery_monitor_is_active(&self) -> bool {
+        self.recovery_monitor_active.load(Ordering::Acquire)
+    }
+
+    #[cfg(test)]
+    fn recovery_monitor_start_count(&self) -> u32 {
+        self.recovery_monitor_start_count.load(Ordering::Acquire)
+    }
+
     pub fn offline_duration_secs(&self) -> Option<u64> {
         self.health.offline_duration().map(|duration| duration.as_secs())
     }
@@ -987,13 +1035,62 @@ 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),
+            #[cfg(test)]
+            RecoveryMonitorTestState {
+                start_count: Arc::clone(&self.recovery_monitor_start_count),
+                teardown_hook: Arc::clone(&self.recovery_monitor_teardown_hook),
+            },
+        );
+    }
+
+    fn schedule_recovery_monitor(
+        addr: String,
+        endpoint: Endpoint,
+        handle_id: Uuid,
+        health: Arc<DiskHealthTracker>,
+        cancel_token: CancellationToken,
+        active: Arc<AtomicBool>,
+        #[cfg(test)] test_state: RecoveryMonitorTestState,
+    ) {
+        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;
+            #[cfg(test)]
+            test_state.start_count.fetch_add(1, Ordering::AcqRel);
+            let lease = RecoveryMonitorLease {
+                active: Arc::clone(&active),
+            };
+            Self::monitor_remote_disk_recovery(addr.clone(), endpoint.clone(), Arc::clone(&health), cancel_token.clone()).await;
+            #[cfg(test)]
+            if let Some(hook) = test_state.teardown_hook.lock().await.take() {
+                hook.arrived.notify_one();
+                hook.release.notified().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,
+                    #[cfg(test)]
+                    test_state,
+                );
+            }
         });
     }
 
@@ -1002,7 +1099,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,
@@ -1033,9 +1130,23 @@ 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);
+        #[cfg(test)]
+        let recovery_monitor_teardown_hook = Arc::clone(&self.recovery_monitor_teardown_hook);
 
         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,
+                #[cfg(test)]
+                recovery_monitor_teardown_hook,
+            )
+            .await;
         });
     }
 
@@ -1043,8 +1154,11 @@ 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>,
+        #[cfg(test)] recovery_monitor_teardown_hook: Arc<tokio::sync::Mutex<Option<Arc<RecoveryMonitorTeardownHook>>>>,
     ) {
         let mut interval = time::interval(get_drive_active_check_interval());
 
@@ -1069,11 +1183,19 @@ 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),
+                #[cfg(test)]
+                RecoveryMonitorTestState {
+                    start_count: Arc::new(AtomicU32::new(0)),
+                    teardown_hook: Arc::clone(&recovery_monitor_teardown_hook),
+                },
+            );
         }
 
         loop {
@@ -1132,11 +1254,19 @@ 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),
+                            #[cfg(test)]
+                            RecoveryMonitorTestState {
+                                start_count: Arc::new(AtomicU32::new(0)),
+                                teardown_hook: Arc::clone(&recovery_monitor_teardown_hook),
+                            },
+                        );
                     }
                 }
             }
@@ -1387,6 +1517,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"
@@ -3548,15 +3679,23 @@ mod tests {
     use crate::erasure::coding::{BitrotReader, Erasure, decode::ParallelReader};
     use crate::io_support::bitrot::ShardReader;
     use crate::runtime::sources as runtime_sources;
+    use rustfs_protos::proto_gen::node_service::{DiskInfoResponse, ReadAllResponse};
     use serde_json::Value;
     use serial_test::serial;
+    use std::convert::Infallible;
+    use std::future::Future;
     use std::io::{self as std_io, Write};
     use std::pin::Pin;
     use std::sync::{Arc, Mutex, Mutex as StdMutex, Once, atomic::AtomicUsize};
     use std::task::{Context, Poll};
     use tokio::io::{ReadBuf, duplex};
     use tokio::net::TcpListener;
-    use tonic::transport::Endpoint as TonicEndpoint;
+    use tonic::transport::{Endpoint as TonicEndpoint, Server};
+    use tonic::{Response, Status};
+    use tonic::{
+        codegen::{Body as HttpBody, BoxFuture, StdError, http},
+        server::NamedService,
+    };
     use tracing::Level;
     use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};
     use uuid::Uuid;
@@ -3716,6 +3855,218 @@ mod tests {
         ns_scanner_probe_status: Arc<StdMutex<Option<u16>>>,
     }
 
+    #[derive(Clone, Debug)]
+    struct AuthenticatedReadPeer {
+        audience: String,
+        disk_info_calls: Arc<AtomicU32>,
+        read_all_calls: Arc<AtomicU32>,
+        object_read_all_disks: Arc<StdMutex<Vec<String>>>,
+        format_data: Bytes,
+        read_all_data: Bytes,
+    }
+
+    impl AuthenticatedReadPeer {
+        fn new(audience: String, format_data: Bytes, read_all_data: Bytes) -> Self {
+            Self {
+                audience,
+                disk_info_calls: Arc::new(AtomicU32::new(0)),
+                read_all_calls: Arc::new(AtomicU32::new(0)),
+                object_read_all_disks: Arc::default(),
+                format_data,
+                read_all_data,
+            }
+        }
+
+        fn disk_info_calls(&self) -> u32 {
+            self.disk_info_calls.load(Ordering::Acquire)
+        }
+
+        fn read_all_calls(&self) -> u32 {
+            self.read_all_calls.load(Ordering::Acquire)
+        }
+
+        fn object_read_all_disks(&self) -> Vec<String> {
+            self.object_read_all_disks
+                .lock()
+                .expect("object read_all disk list lock poisoned")
+                .clone()
+        }
+
+        fn verify_auth<T>(&self, request: &Request<T>, path: &str) -> std::result::Result<(), Status> {
+            let headers = request.metadata().clone().into_headers();
+            crate::cluster::rpc::verify_tonic_rpc_signature(&self.audience, path, &headers)
+                .map_err(|err| Status::unauthenticated(err.to_string()))
+        }
+    }
+
+    #[derive(Clone, Debug)]
+    struct AuthenticatedReadPeerService {
+        peer: AuthenticatedReadPeer,
+    }
+
+    impl NamedService for AuthenticatedReadPeerService {
+        const NAME: &'static str = "node_service.NodeService";
+    }
+
+    impl<B> tower::Service<http::Request<B>> for AuthenticatedReadPeerService
+    where
+        B: HttpBody + Send + 'static,
+        B::Error: Into<StdError> + Send + 'static,
+    {
+        type Response = http::Response<tonic::body::Body>;
+        type Error = Infallible;
+        type Future = BoxFuture<Self::Response, Self::Error>;
+
+        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn call(&mut self, request: http::Request<B>) -> Self::Future {
+            match request.uri().path() {
+                "/node_service.NodeService/DiskInfo" => {
+                    #[derive(Clone)]
+                    struct DiskInfoSvc(AuthenticatedReadPeer);
+
+                    impl tonic::server::UnaryService<DiskInfoRequest> for DiskInfoSvc {
+                        type Response = DiskInfoResponse;
+                        type Future = Pin<Box<dyn Future<Output = std::result::Result<Response<Self::Response>, Status>> + Send>>;
+
+                        fn call(&mut self, request: Request<DiskInfoRequest>) -> Self::Future {
+                            let peer = self.0.clone();
+                            Box::pin(async move {
+                                peer.verify_auth(&request, "/node_service.NodeService/DiskInfo")?;
+                                let request = request.into_inner();
+                                let opts = serde_json::from_str::<DiskInfoOptions>(&request.opts)
+                                    .map_err(|err| Status::invalid_argument(err.to_string()))?;
+                                if !opts.noop {
+                                    return Err(Status::invalid_argument("recovery probe must use noop disk_info"));
+                                }
+                                peer.disk_info_calls.fetch_add(1, Ordering::AcqRel);
+                                let disk_info = serde_json::to_string(&DiskInfo {
+                                    total: 1,
+                                    free: 1,
+                                    endpoint: request.disk,
+                                    ..Default::default()
+                                })
+                                .map_err(|err| Status::internal(err.to_string()))?;
+                                Ok(Response::new(DiskInfoResponse {
+                                    success: true,
+                                    disk_info,
+                                    error: None,
+                                }))
+                            })
+                        }
+                    }
+
+                    let peer = self.peer.clone();
+                    Box::pin(async move {
+                        let method = DiskInfoSvc(peer);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec);
+                        Ok(grpc.unary(method, request).await)
+                    })
+                }
+                "/node_service.NodeService/ReadAll" => {
+                    #[derive(Clone)]
+                    struct ReadAllSvc(AuthenticatedReadPeer);
+
+                    impl tonic::server::UnaryService<ReadAllRequest> for ReadAllSvc {
+                        type Response = ReadAllResponse;
+                        type Future = Pin<Box<dyn Future<Output = std::result::Result<Response<Self::Response>, Status>> + Send>>;
+
+                        fn call(&mut self, request: Request<ReadAllRequest>) -> Self::Future {
+                            let peer = self.0.clone();
+                            Box::pin(async move {
+                                peer.verify_auth(&request, "/node_service.NodeService/ReadAll")?;
+                                let request = request.into_inner();
+                                let is_format_read = request.volume == crate::disk::RUSTFS_META_BUCKET
+                                    && request.path == crate::disk::FORMAT_CONFIG_FILE;
+                                let disk = request.disk;
+                                peer.read_all_calls.fetch_add(1, Ordering::AcqRel);
+                                let data = if is_format_read {
+                                    peer.format_data.clone()
+                                } else {
+                                    peer.object_read_all_disks
+                                        .lock()
+                                        .expect("object read_all disk list lock poisoned")
+                                        .push(disk);
+                                    peer.read_all_data.clone()
+                                };
+                                Ok(Response::new(ReadAllResponse {
+                                    success: true,
+                                    data,
+                                    error: None,
+                                }))
+                            })
+                        }
+                    }
+
+                    let peer = self.peer.clone();
+                    Box::pin(async move {
+                        let method = ReadAllSvc(peer);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec);
+                        Ok(grpc.unary(method, request).await)
+                    })
+                }
+                _ => Box::pin(async move {
+                    let mut response = http::Response::new(tonic::body::Body::default());
+                    let headers = response.headers_mut();
+                    headers.insert(tonic::Status::GRPC_STATUS, (tonic::Code::Unimplemented as i32).into());
+                    headers.insert(http::header::CONTENT_TYPE, tonic::metadata::GRPC_CONTENT_TYPE);
+                    Ok(response)
+                }),
+            }
+        }
+    }
+
+    struct TestGrpcPeer {
+        addr: String,
+        peer: AuthenticatedReadPeer,
+        shutdown: CancellationToken,
+        task: tokio::task::JoinHandle<()>,
+    }
+
+    impl TestGrpcPeer {
+        async fn spawn(format_data: Bytes, read_all_data: Bytes) -> Option<Self> {
+            let listener = match TcpListener::bind("127.0.0.1:0").await {
+                Ok(listener) => listener,
+                Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
+                Err(err) => panic!("test gRPC listener should bind: {err}"),
+            };
+            let socket_addr = listener.local_addr().expect("listener local address should be available");
+            let addr = format!("http://{socket_addr}");
+            let audience = crate::cluster::rpc::normalize_tonic_rpc_audience(&socket_addr.to_string())
+                .expect("test audience should normalize");
+            let peer = AuthenticatedReadPeer::new(audience, format_data, read_all_data);
+            let service = AuthenticatedReadPeerService { peer: peer.clone() };
+            let shutdown = CancellationToken::new();
+            let shutdown_for_task = shutdown.clone();
+            let incoming = futures_util::stream::unfold(listener, |listener| async {
+                Some((listener.accept().await.map(|(stream, _)| stream), listener))
+            });
+            let task = tokio::spawn(async move {
+                Server::builder()
+                    .add_service(service)
+                    .serve_with_incoming_shutdown(incoming, shutdown_for_task.cancelled_owned())
+                    .await
+                    .expect("test gRPC peer should serve");
+            });
+
+            Some(Self {
+                addr,
+                peer,
+                shutdown,
+                task,
+            })
+        }
+
+        async fn stop(self) {
+            self.shutdown.cancel();
+            let _ = self.task.await;
+        }
+    }
+
     impl RecordingInternodeDataTransport {
         fn with_ns_scanner_probe_status(status: u16) -> Self {
             Self {
@@ -5558,6 +5909,245 @@ 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();
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while disk.recovery_monitor_start_count() == 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("recovery monitor should start");
+
+        assert!(disk.recovery_monitor_is_active(), "only one recovery monitor should own the handle");
+        assert_eq!(
+            disk.recovery_monitor_start_count(),
+            1,
+            "the failed compare-exchange path must not start a second monitor"
+        );
+
+        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]
+    #[serial(remote_disk_recovery_probe)]
+    async fn recovery_monitor_rearms_if_disk_fails_during_teardown() {
+        runtime_sources::ensure_test_rpc_secret();
+        let Some(peer) = TestGrpcPeer::spawn(Bytes::new(), Bytes::new()).await else {
+            return;
+        };
+        let endpoint = Endpoint {
+            url: url::Url::parse(&format!("{}/data/rustfs0", peer.addr)).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 {
+            peer.stop().await;
+            return;
+        }
+
+        disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+        let hook = Arc::new(RecoveryMonitorTeardownHook::default());
+        *disk.recovery_monitor_teardown_hook.lock().await = Some(Arc::clone(&hook));
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_DRIVE_RETURNING_PROBE_INTERVAL_SECS, Some("1")),
+                (rustfs_config::ENV_DRIVE_RETURNING_SUCCESS_THRESHOLD, Some("1")),
+                (rustfs_config::ENV_DRIVE_ACTIVE_CHECK_TIMEOUT_SECS, Some("1")),
+            ],
+            async {
+                disk.spawn_recovery_monitor_if_needed();
+                tokio::time::timeout(Duration::from_secs(5), hook.arrived.notified())
+                    .await
+                    .expect("first recovery monitor should reach teardown");
+                assert_eq!(disk.runtime_state(), RuntimeDriveHealthState::Online);
+
+                disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+                hook.release.notify_one();
+                tokio::time::timeout(Duration::from_secs(2), async {
+                    while disk.recovery_monitor_start_count() < 2 {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .expect("teardown failure should re-arm recovery monitoring");
+                assert!(
+                    disk.recovery_monitor_is_active(),
+                    "re-armed monitor should retain single-flight ownership"
+                );
+
+                disk.cancel_token.cancel();
+                tokio::time::timeout(Duration::from_secs(2), async {
+                    while disk.recovery_monitor_is_active() {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .expect("cancelled re-armed monitor should release single-flight state");
+            },
+        )
+        .await;
+
+        peer.stop().await;
+    }
+
+    #[tokio::test]
+    #[serial(remote_disk_recovery_probe)]
+    async fn recovery_monitor_restores_online_then_real_reads_use_replacement_handle() {
+        runtime_sources::ensure_test_rpc_secret();
+        let mut format = crate::layout::format::FormatV3::new(1, 1);
+        let disk_id = format.erasure.sets[0][0];
+        format.erasure.this = disk_id;
+        let format_data = Bytes::from(format.to_json().expect("test format should serialize"));
+        let Some(peer) = TestGrpcPeer::spawn(format_data, Bytes::from_static(b"replacement-data")).await else {
+            return;
+        };
+        let url = url::Url::parse(&format!("{}/data/rustfs0", peer.addr)).expect("endpoint should parse");
+        let endpoint = Endpoint {
+            url,
+            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");
+        disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_DRIVE_RETURNING_PROBE_INTERVAL_SECS, Some("1")),
+                (rustfs_config::ENV_DRIVE_RETURNING_SUCCESS_THRESHOLD, Some("3")),
+                (rustfs_config::ENV_DRIVE_ACTIVE_CHECK_TIMEOUT_SECS, Some("1")),
+            ],
+            async {
+                let monitor = tokio::spawn(RemoteDisk::monitor_remote_disk_recovery(
+                    disk.addr.clone(),
+                    endpoint.clone(),
+                    Arc::clone(&disk.health),
+                    disk.cancel_token.clone(),
+                ));
+
+                tokio::time::timeout(Duration::from_secs(5), async {
+                    while disk.runtime_state() != RuntimeDriveHealthState::Online {
+                        tokio::time::sleep(Duration::from_millis(20)).await;
+                    }
+                })
+                .await
+                .expect("three authenticated recovery probes should restore the disk online");
+                monitor.await.expect("recovery monitor should exit after restoring Online");
+
+                assert_eq!(
+                    peer.peer.disk_info_calls(),
+                    3,
+                    "RemoteDisk recovery requires the configured three successful disk_info probes"
+                );
+                let recovered_read = disk.read_all("bucket", "object").await.expect("recovered handle should read");
+                assert_eq!(recovered_read, Bytes::from_static(b"replacement-data"));
+
+                let old_disk = crate::disk::new_disk(
+                    &endpoint,
+                    &DiskOption {
+                        cleanup: false,
+                        health_check: false,
+                    },
+                )
+                .await
+                .expect("old slot disk should construct");
+                let set_disks = crate::set_disk::SetDisks::new(
+                    "remote-recovery-test".to_string(),
+                    Arc::new(tokio::sync::RwLock::new(vec![Some(old_disk.clone())])),
+                    1,
+                    0,
+                    0,
+                    0,
+                    vec![endpoint.clone()],
+                    format,
+                    Vec::new(),
+                )
+                .await;
+                set_disks.disks.write().await[0] = None;
+                set_disks.renew_disk(&endpoint).await;
+
+                let slots = set_disks.disks.read().await;
+                let replacement = slots[0]
+                    .as_ref()
+                    .expect("renew_disk should publish the replacement slot")
+                    .clone();
+                drop(slots);
+                assert!(!Arc::ptr_eq(&replacement, &old_disk), "renew_disk must replace the stale slot handle");
+                let replacement_read = replacement
+                    .read_all("bucket", "object")
+                    .await
+                    .expect("production slot should route real reads through the replacement");
+                assert_eq!(replacement_read, Bytes::from_static(b"replacement-data"));
+                let object_reads = peer.peer.object_read_all_disks();
+                assert_eq!(object_reads.len(), 2, "standalone and production-slot reads should both reach the peer");
+                assert_eq!(object_reads[1], disk_id.to_string(), "production slot must use the renewed disk identity");
+                assert!(peer.peer.read_all_calls() >= 3, "renewal must read format metadata before the slot read");
+                disk.cancel_token.cancel();
+                old_disk.close().await.expect("old slot disk should close");
+                replacement.close().await.expect("replacement slot disk should close");
+            },
+        )
+        .await;
+
+        peer.stop().await;
+    }
+
     #[tokio::test]
     async fn test_copy_stream_with_buffer_copies_full_payload() {
         let payload = b"walk-dir-stream".repeat(1024);
@@ -7124,6 +7714,20 @@ mod tests {
         )
         .await
         .expect("remote disk should construct");
+        let replacement = RemoteDisk::new(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: true,
+            },
+            Arc::new(TcpHttpInternodeDataTransport),
+        )
+        .await
+        .expect("replacement remote disk should construct");
+        assert_ne!(
+            remote_disk.handle_id, replacement.handle_id,
+            "replacement handles need distinct log identities"
+        );
 
         let span = tracing::info_span!("request-span", request_id = "req-remote-disk");
         let _entered = span.enter();
@@ -7139,11 +7743,24 @@ mod tests {
 
         assert_eq!(log["span"]["name"], Value::String("recovery-monitor".to_string()));
         assert_eq!(log["span"]["kind"], Value::String("remote_disk".to_string()));
+        assert_eq!(log["span"]["handle_id"], Value::String(remote_disk.handle_id.to_string()));
         let spans = log["spans"].as_array().expect("spans should be present");
         assert!(spans.iter().any(|span| {
             span.get("name").and_then(Value::as_str) == Some("request-span")
                 && span.get("request_id").and_then(Value::as_str) == Some("req-remote-disk")
         }));
+
+        remote_disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+        remote_disk
+            .execute_with_timeout(|| async { Ok::<(), Error>(()) }, Duration::from_secs(1))
+            .await
+            .expect_err("faulty handle should short-circuit");
+        let faulty_log = logs
+            .lines()
+            .into_iter()
+            .find(|value| value.get("state").and_then(Value::as_str) == Some("faulty_short_circuit"))
+            .expect("expected faulty short-circuit log");
+        assert_eq!(faulty_log["handle_id"], Value::String(remote_disk.handle_id.to_string()));
     }
 
     #[tokio::test(flavor = "current_thread")]

crates/ecstore/src/disk/disk_store.rs

@@ -418,6 +418,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)]
@@ -739,6 +750,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(()),
         }
     }
 
@@ -775,39 +788,52 @@ 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
+    }
+
+    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
     #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     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> {
@@ -823,6 +849,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 {
@@ -851,24 +878,19 @@ 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
     }
 
     #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     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
     }
@@ -882,11 +904,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);
@@ -898,6 +919,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,
@@ -918,7 +940,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);
         }
@@ -948,7 +969,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);
 
@@ -1217,7 +1244,7 @@ impl LocalDiskWrapper {
                         return;
                     }
 
-                    if health.status.load(Ordering::Relaxed) != DISK_HEALTH_OK {
+                    if health.is_faulty() {
                         continue;
                     }
 
@@ -2909,6 +2936,57 @@ mod tests {
         });
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn concurrent_failure_and_recovery_publish_one_health_snapshot() {
+        temp_env::with_var(rustfs_config::ENV_DRIVE_SUSPECT_FAILURE_THRESHOLD, Some("2"), || {
+            let endpoint = Endpoint::try_from("/tmp/concurrent-health-snapshot").expect("endpoint should parse");
+            let health = Arc::new(DiskHealthTracker::new());
+            let transition_guard = health
+                .transition_lock
+                .lock()
+                .expect("health transition lock should not be poisoned");
+            let start = Arc::new(std::sync::Barrier::new(3));
+            let (completed_tx, completed_rx) = std::sync::mpsc::channel();
+            let workers = (0..2)
+                .map(|_| {
+                    let health = Arc::clone(&health);
+                    let endpoint = endpoint.clone();
+                    let start = Arc::clone(&start);
+                    let completed_tx = completed_tx.clone();
+                    std::thread::spawn(move || {
+                        start.wait();
+                        health.mark_failure(&endpoint, "concurrent_test");
+                        completed_tx.send(()).expect("completion receiver should remain available");
+                    })
+                })
+                .collect::<Vec<_>>();
+
+            start.wait();
+            assert!(
+                matches!(
+                    completed_rx.recv_timeout(Duration::from_millis(250)),
+                    Err(std::sync::mpsc::RecvTimeoutError::Timeout)
+                ),
+                "concurrent transitions must wait for the serialization lock"
+            );
+            drop(transition_guard);
+            completed_rx
+                .recv_timeout(Duration::from_secs(1))
+                .expect("first failure transition should complete after lock release");
+            completed_rx
+                .recv_timeout(Duration::from_secs(1))
+                .expect("second failure transition should complete after lock release");
+            for worker in workers {
+                worker.join().expect("health transition worker should not panic");
+            }
+
+            assert_eq!(health.runtime_state(), RuntimeDriveHealthState::Offline);
+            assert!(health.is_faulty());
+            assert_eq!(health.consecutive_failures.load(Ordering::Acquire), 2);
+        });
+    }
+
     #[test]
     fn operation_success_recovers_suspect_drive_without_faulting() {
         let endpoint = Endpoint::try_from("/tmp/runtime-state-suspect-success").expect("endpoint should parse");

crates/protos/src/lib.rs

@@ -2557,7 +2557,7 @@ mod tests {
 
     fn production_source(source: &'static str, file_name: &str) -> &'static str {
         source
-            .split("\n#[cfg(test)]")
+            .split("\n#[cfg(test)]\nmod tests")
             .next()
             .unwrap_or_else(|| panic!("{file_name} should contain production source before tests"))
     }