From a4ea36b298b5e1713c29178f03f8b6bdc796a761 Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 18 Aug 2026 14:49:53 +0800 Subject: [PATCH] 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 | 1 + crates/ecstore/Cargo.toml | 1 + crates/ecstore/src/cluster/rpc/remote_disk.rs | 661 +++++++++++++++++- crates/ecstore/src/disk/disk_store.rs | 130 +++- crates/protos/src/lib.rs | 2 +- 5 files changed, 746 insertions(+), 49 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 82ffafc81..b297135af 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9465,6 +9465,7 @@ dependencies = [ "tokio-stream", "tokio-util", "tonic", + "tonic-prost", "tower", "tracing", "tracing-core", diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index bdbb53e49..6525f0cfd 100644 --- a/crates/ecstore/Cargo.toml +++ b/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"] } diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 3ae2c8b01..26e9f899a 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/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>, pub addr: String, endpoint: Endpoint, @@ -640,9 +642,38 @@ pub struct RemoteDisk { health: Arc, /// Cancellation token for monitoring tasks cancel_token: CancellationToken, + recovery_monitor_active: Arc, + #[cfg(test)] + recovery_monitor_start_count: Arc, + #[cfg(test)] + recovery_monitor_teardown_hook: Arc>>>, data_transport: Arc, } +struct RecoveryMonitorLease { + active: Arc, +} + +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, + teardown_hook: Arc>>>, +} + // ── 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 { self.health.offline_duration().map(|duration| duration.as_secs()) } @@ -987,13 +1035,62 @@ 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), + #[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, + cancel_token: CancellationToken, + active: Arc, + #[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, cancel_token: CancellationToken, + recovery_monitor_active: Arc, + #[cfg(test)] recovery_monitor_teardown_hook: Arc>>>, ) { 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); - - 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), + #[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); - - 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), + #[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>>, } + #[derive(Clone, Debug)] + struct AuthenticatedReadPeer { + audience: String, + disk_info_calls: Arc, + read_all_calls: Arc, + object_read_all_disks: Arc>>, + 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 { + self.object_read_all_disks + .lock() + .expect("object read_all disk list lock poisoned") + .clone() + } + + fn verify_auth(&self, request: &Request, 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 tower::Service> for AuthenticatedReadPeerService + where + B: HttpBody + Send + 'static, + B::Error: Into + Send + 'static, + { + type Response = http::Response; + type Error = Infallible; + type Future = BoxFuture; + + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, request: http::Request) -> Self::Future { + match request.uri().path() { + "/node_service.NodeService/DiskInfo" => { + #[derive(Clone)] + struct DiskInfoSvc(AuthenticatedReadPeer); + + impl tonic::server::UnaryService for DiskInfoSvc { + type Response = DiskInfoResponse; + type Future = Pin, Status>> + Send>>; + + fn call(&mut self, request: Request) -> 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::(&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 for ReadAllSvc { + type Response = ReadAllResponse; + type Future = Pin, Status>> + Send>>; + + fn call(&mut self, request: Request) -> 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 { + 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")] diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 29a8037dd..5c0f6454e 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/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); - 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 { @@ -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.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); @@ -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::>(); + + 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"); diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index a826d4a31..d23224b93 100644 --- a/crates/protos/src/lib.rs +++ b/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")) }