diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index e519585e4..bc206d702 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -5745,6 +5745,7 @@ mod tests { object_read_all_disks: Arc>>, format_data: Bytes, read_all_data: Bytes, + fail_format_reads: Arc, next_format_read_gate: Arc>>>, unsupported_methods: Arc>>, } @@ -5764,6 +5765,7 @@ mod tests { object_read_all_disks: Arc::default(), format_data, read_all_data, + fail_format_reads: Arc::default(), next_format_read_gate: Arc::default(), unsupported_methods: Arc::default(), } @@ -5887,6 +5889,9 @@ mod tests { gate.entered.notify_one(); gate.release.cancelled().await; } + if peer.fail_format_reads.load(Ordering::Acquire) { + return Err(Status::unavailable("format read temporarily unavailable")); + } peer.format_data.clone() } else { peer.object_read_all_disks @@ -8852,6 +8857,124 @@ mod tests { peer.stop().await; } + #[tokio::test] + #[serial(remote_disk_recovery_probe)] + async fn endpoint_reconnect_failure_preserves_registered_remote_recovery() { + temp_env::async_with_vars( + [ + (crate::disk::disk_store::ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, Some("true")), + (rustfs_config::ENV_DRIVE_RETURNING_SUCCESS_THRESHOLD, Some("1")), + ], + async { + runtime_sources::ensure_test_rpc_secret(); + let mut reference = crate::layout::format::FormatV3::new(1, 1); + reference.erasure.this = reference.erasure.sets[0][0]; + let peer = TestGrpcPeer::spawn( + Bytes::from(reference.to_json().expect("format should serialize")), + Bytes::from_static(b"recovered-peer-data"), + ) + .await + .expect("authenticated regression peer is required"); + let set = missing_read_set(&peer, reference).await; + let disk = crate::disk::new_disk( + &set.set_endpoints[0], + &DiskOption { + cleanup: false, + health_check: true, + }, + ) + .await + .expect("registered remote disk should construct"); + let crate::disk::Disk::Remote(remote) = disk.as_ref() else { + panic!("fixture must create a remote disk"); + }; + assert!(remote.health_check); + remote.force_runtime_state_for_test(RuntimeDriveHealthState::Offline); + set.disks.write().await[0] = Some(Arc::clone(&disk)); + + peer.peer.fail_format_reads.store(true, Ordering::Release); + set.connect_disks().await; + peer.peer.fail_format_reads.store(false, Ordering::Release); + let cancelled_after_failure = remote.cancel_token.is_cancelled(); + assert!(peer.peer.read_all_calls() > 0, "reconnect must attempt the failing format RPC"); + assert!( + Arc::ptr_eq(&disk, set.disks.read().await[0].as_ref().expect("existing slot must remain")), + "a failed reconnect must not publish its unvalidated candidate" + ); + remote.spawn_recovery_monitor_if_needed(); + let recovery = time::timeout(Duration::from_secs(5), async { + while !disk.is_online().await { + tokio::task::yield_now().await; + } + disk.read_all("bucket", "object").await + }) + .await; + disk.close().await.expect("stop the registered disk's monitoring"); + peer.stop().await; + assert!(!cancelled_after_failure, "failed replacement must retain the registered monitor"); + assert_eq!( + recovery + .expect("failed reconnect must leave the registered disk able to recover without another sweep") + .expect("the recovered registered disk must serve real I/O"), + Bytes::from_static(b"recovered-peer-data") + ); + }, + ) + .await; + } + + #[tokio::test] + #[serial(remote_disk_recovery_probe)] + async fn endpoint_reconnect_stops_old_remote_monitor_only_after_publication() { + runtime_sources::ensure_test_rpc_secret(); + let mut reference = crate::layout::format::FormatV3::new(1, 1); + reference.erasure.this = reference.erasure.sets[0][0]; + let peer = TestGrpcPeer::spawn( + Bytes::from(reference.to_json().expect("format should serialize")), + Bytes::from_static(b"replacement-peer-data"), + ) + .await + .expect("authenticated regression peer is required"); + let set = missing_read_set(&peer, reference).await; + let disk = crate::disk::new_disk( + &set.set_endpoints[0], + &DiskOption { + cleanup: false, + health_check: true, + }, + ) + .await + .expect("registered remote disk should construct"); + let crate::disk::Disk::Remote(remote) = disk.as_ref() else { + panic!("fixture must create a remote disk"); + }; + remote.force_runtime_state_for_test(RuntimeDriveHealthState::Offline); + set.disks.write().await[0] = Some(Arc::clone(&disk)); + let gate = peer.peer.pause_next_format_read(); + let pending_set = Arc::clone(&set); + let pending = tokio::spawn(async move { pending_set.connect_disks().await }); + time::timeout(Duration::from_secs(5), gate.entered.notified()) + .await + .expect("replacement must reach the format validation barrier"); + let cancelled_before_publication = remote.cancel_token.is_cancelled(); + gate.release.cancel(); + pending.await.expect("validated replacement must finish publication"); + let replacement = set.disks.read().await[0].clone().expect("replacement must occupy the slot"); + let readback = replacement.read_all("bucket", "object").await; + replacement.close().await.expect("stop replacement monitoring"); + peer.stop().await; + assert!( + !cancelled_before_publication, + "the registered monitor must survive pending format validation" + ); + assert!(!Arc::ptr_eq(&disk, &replacement), "validated replacement must be published"); + assert!(remote.cancel_token.is_cancelled(), "the displaced monitor must stop after publication"); + assert_eq!( + readback.expect("replacement must serve real I/O"), + Bytes::from_static(b"replacement-peer-data") + ); + } + #[tokio::test] #[serial(remote_disk_recovery_probe)] async fn endpoint_reconnect_admits_ready_drive_while_another_format_probe_is_blocked() { diff --git a/crates/ecstore/src/set_disk/ops/locking.rs b/crates/ecstore/src/set_disk/ops/locking.rs index 905e66598..b6c656fe7 100644 --- a/crates/ecstore/src/set_disk/ops/locking.rs +++ b/crates/ecstore/src/set_disk/ops/locking.rs @@ -375,7 +375,10 @@ impl SetDisks { return None; } - let _ = disk.close().await; + // A failed remote reconnect must retain its existing recovery monitor. + if disk.is_local() { + let _ = disk.close().await; + } } let endpoint = self.set_endpoints.get(i)?; @@ -497,19 +500,26 @@ impl SetDisks { let _ = new_disk.set_disk_id(Some(fm.erasure.this)).await; if !new_disk.is_local() { - let mut disks = self.disks.write().await; - let slot = &mut disks[disk_idx]; - let unchanged = match (slot.as_ref(), previous_disk.as_ref()) { - (None, None) => true, - (Some(current), Some(previous)) => Arc::ptr_eq(current, previous), - _ => false, + let displaced = { + let mut disks = self.disks.write().await; + let slot = &mut disks[disk_idx]; + let unchanged = match (slot.as_ref(), previous_disk.as_ref()) { + (None, None) => true, + (Some(current), Some(previous)) => Arc::ptr_eq(current, previous), + _ => false, + }; + // A GET probe or another reconnect may have filled this slot while + // format I/O was pending. Only the successful publisher owns the + // displaced monitor; a failed or superseded probe must retain it. + if unchanged { + new_disk.enable_health_check(); + slot.replace(new_disk) + } else { + None + } }; - // A GET probe or another reconnect may have filled this slot while - // format I/O was pending. Do not replace its handle or start a - // second health monitor for an unpublished handle. - if unchanged { - new_disk.enable_health_check(); - *slot = Some(new_disk); + if let Some(disk) = displaced { + let _ = disk.close().await; } return false; }