diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 456d38ac7..73468a2c2 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -454,13 +454,14 @@ pub mod rpc { AuthenticatedChannel, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client, S3PeerSys, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerBucketListing, ScannerPeerActivity, TONIC_RPC_PREFIX, TonicInterceptor, build_put_file_auth_trailer, - check_and_record_signed_rpc_nonce, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers, - gen_tonic_signature_interceptor, node_service_time_out_client, node_service_time_out_client_no_auth, - normalize_tonic_rpc_audience, set_tonic_canonical_body_digest, sign_ns_scanner_capability, sign_put_file_capability, - sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, - tonic_rpc_auth_failure_reason, verify_put_file_auth_trailer, verify_put_file_capability, verify_rpc_signature, - verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, - verify_tonic_rpc_response_proof, verify_tonic_rpc_signature, verify_tonic_rpc_signature_with_bootstrap, + check_and_record_signed_rpc_nonce, decode_heal_bucket_rpc_options, encode_heal_bucket_rpc_options, gen_signature_headers, + gen_tonic_replay_scope_headers, gen_tonic_signature_headers, gen_tonic_signature_interceptor, + node_service_time_out_client, node_service_time_out_client_no_auth, normalize_tonic_rpc_audience, + set_tonic_canonical_body_digest, sign_ns_scanner_capability, sign_put_file_capability, sign_tonic_rpc_response_proof, + tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, + verify_put_file_auth_trailer, verify_put_file_capability, verify_rpc_signature, verify_tonic_boot_epoch_response, + verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof, + verify_tonic_rpc_signature, verify_tonic_rpc_signature_with_bootstrap, }; } diff --git a/crates/ecstore/src/cluster/rpc/mod.rs b/crates/ecstore/src/cluster/rpc/mod.rs index 9af454ac4..4fc826b20 100644 --- a/crates/ecstore/src/cluster/rpc/mod.rs +++ b/crates/ecstore/src/cluster/rpc/mod.rs @@ -50,6 +50,9 @@ pub use peer_rest_client::{ SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerPeerActivity, }; pub(crate) use peer_s3_client::heal_bucket_local_on_disks; -pub use peer_s3_client::{LocalPeerS3Client, PeerS3Client, S3PeerSys, ScannerBucketListing, ScannerSetBucketListing}; +pub use peer_s3_client::{ + LocalPeerS3Client, PeerS3Client, S3PeerSys, ScannerBucketListing, ScannerSetBucketListing, decode_heal_bucket_rpc_options, + encode_heal_bucket_rpc_options, +}; pub use remote_disk::RemoteDisk; pub use remote_locker::RemoteClient; diff --git a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs index 72db2062c..4acff0933 100644 --- a/crates/ecstore/src/cluster/rpc/peer_s3_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_s3_client.rs @@ -18,6 +18,7 @@ use crate::cluster::rpc::client::{ node_service_time_out_client, }; use crate::cluster::rpc::set_tonic_mutation_body_digest; +use crate::core::pools::PoolMeta; use crate::disk::error::DiskError; use crate::disk::error::{Error, Result}; use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs}; @@ -46,7 +47,12 @@ use std::sync::{ Mutex as StdMutex, atomic::{AtomicBool, Ordering}, }; -use std::{collections::HashMap, fmt::Debug, sync::Arc, time::Duration}; +use std::{ + collections::{BTreeSet, HashMap}, + fmt::Debug, + sync::Arc, + time::Duration, +}; #[cfg(test)] use tokio::sync::Notify; use tokio::{net::TcpStream, sync::RwLock, time}; @@ -99,6 +105,9 @@ impl DeleteBucketEmptyScanBarrier { #[cfg(test)] static DELETE_BUCKET_EMPTY_SCAN_BARRIER: StdMutex>> = StdMutex::new(None); +#[cfg(test)] +static HEAL_BUCKET_PRE_MUTATION_BARRIER: StdMutex>> = StdMutex::new(None); + #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] enum HealBucketOperation { Make, @@ -171,6 +180,15 @@ pub(crate) fn install_delete_bucket_empty_scan_barrier() -> Arc Arc { + let barrier = Arc::new(DeleteBucketEmptyScanBarrier::default()); + *HEAL_BUCKET_PRE_MUTATION_BARRIER + .lock() + .expect("heal bucket mutation barrier lock should not be poisoned") = Some(barrier.clone()); + barrier +} + #[cfg(test)] async fn pause_after_delete_bucket_empty_scan() { let barrier = DELETE_BUCKET_EMPTY_SCAN_BARRIER @@ -182,6 +200,20 @@ async fn pause_after_delete_bucket_empty_scan() { } } +#[cfg(test)] +async fn pause_before_heal_bucket_volume_mutation() { + let barrier = HEAL_BUCKET_PRE_MUTATION_BARRIER + .lock() + .expect("heal bucket mutation barrier lock should not be poisoned") + .take(); + if let Some(barrier) = barrier { + barrier.pause().await; + } +} + +#[cfg(not(test))] +async fn pause_before_heal_bucket_volume_mutation() {} + #[derive(Clone, Debug)] pub struct ScannerBucketListing { pub buckets: Vec, @@ -253,9 +285,41 @@ fn resolve_heal_bucket_mode(opts: &mut HealOpts, pool_errs: &[Option]) -> Ok(()) } +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct HealBucketRpcEnvelope { + options: HealOpts, + #[serde(rename = "fencedPools", default)] + fenced_pools: Vec, +} + +pub fn encode_heal_bucket_rpc_options(opts: HealOpts, fenced_pools: &[usize]) -> Result { + serde_json::to_string(&HealBucketRpcEnvelope { + options: opts, + fenced_pools: fenced_pools.to_vec(), + }) + .map_err(Into::into) +} + +pub fn decode_heal_bucket_rpc_options(payload: &str) -> Result<(HealOpts, Vec)> { + match serde_json::from_str::(payload) { + Ok(envelope) => Ok((envelope.options, envelope.fenced_pools)), + Err(envelope_err) => serde_json::from_str::(payload) + .map(|options| (options, Vec::new())) + .map_err(|legacy_err| { + Error::other(format!( + "decode heal bucket RPC options failed: envelope={envelope_err}; legacy={legacy_err}" + )) + }), + } +} + #[async_trait] pub trait PeerS3Client: Debug + Sync + Send + 'static { async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result; + async fn heal_bucket_with_fence(&self, bucket: &str, opts: &HealOpts, _fenced_pools: &[usize]) -> Result { + self.heal_bucket(bucket, opts).await + } async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>; async fn list_bucket(&self, opts: &BucketOptions) -> Result>; async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()>; @@ -309,6 +373,10 @@ impl S3PeerSys { impl S3PeerSys { pub async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result { + self.heal_bucket_with_fence(bucket, opts, &[]).await + } + + pub async fn heal_bucket_with_fence(&self, bucket: &str, opts: &HealOpts, fenced_pools: &[usize]) -> Result { let mut opts = *opts; let mut futures = Vec::with_capacity(self.clients.len()); for client in self.clients.iter() { @@ -331,7 +399,7 @@ impl S3PeerSys { let opts_clone = opts; let heal_bucket_results_clone = heal_bucket_results.clone(); futures.push(async move { - match client.heal_bucket(bucket, &opts_clone).await { + match client.heal_bucket_with_fence(bucket, &opts_clone, fenced_pools).await { Ok(res) => { heal_bucket_results_clone.write().await[idx] = res; None @@ -635,8 +703,18 @@ impl PeerS3Client for LocalPeerS3Client { } async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result { + self.heal_bucket_with_fence(bucket, opts, &[]).await + } + + async fn heal_bucket_with_fence(&self, bucket: &str, opts: &HealOpts, fenced_pools: &[usize]) -> Result { let disks = self.local_disks_for_pools().await.into_iter().map(Some).collect(); - heal_bucket_local_on_disks(bucket, opts, disks).await + let store = runtime_sources::object_store_handle().filter(|store| Arc::ptr_eq(&store.ctx, &self.instance_ctx)); + #[cfg(not(test))] + if store.is_none() { + return Err(Error::other("bucket heal refused: pool metadata is unavailable for this instance")); + } + heal_bucket_local_on_disks_with_pool_meta(bucket, opts, disks, store.as_ref().map(|store| &store.pool_meta), fenced_pools) + .await } async fn list_bucket(&self, _opts: &BucketOptions) -> Result> { @@ -1079,9 +1157,13 @@ impl PeerS3Client for RemotePeerS3Client { } async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result { + self.heal_bucket_with_fence(bucket, opts, &[]).await + } + + async fn heal_bucket_with_fence(&self, bucket: &str, opts: &HealOpts, fenced_pools: &[usize]) -> Result { self.execute_with_timeout( || async { - let options: String = serde_json::to_string(opts)?; + let options = encode_heal_bucket_rpc_options(*opts, fenced_pools)?; let mut client = self.get_client().await?; let mut request = Request::new(HealBucketRequest { bucket: bucket.to_string(), @@ -1229,6 +1311,117 @@ pub(crate) async fn heal_bucket_local_on_disks( opts: &HealOpts, disks: Vec>, ) -> Result { + if let Some(store) = runtime_sources::object_store_handle() { + return heal_bucket_local_on_disks_with_pool_meta(bucket, opts, disks, Some(&store.pool_meta), &[]).await; + } + + #[cfg(test)] + return heal_bucket_local_on_disks_with_pool_meta(bucket, opts, disks, None, &[]).await; + + #[cfg(not(test))] + Err(Error::other("bucket heal refused: pool metadata is unavailable")) +} + +fn disk_pool_index(disk: &DiskStore) -> Result { + usize::try_from(disk.endpoint().pool_idx) + .map_err(|_| Error::other(format!("invalid bucket-heal pool index {}", disk.endpoint().pool_idx))) +} + +fn fenced_decommission_drive_state() -> DriveState { + DriveState::Unknown("skipped-decommission-suspended".to_string()) +} + +fn heal_bucket_fence_detail(fenced_pools: &BTreeSet) -> Option { + if fenced_pools.is_empty() { + return None; + } + let pools = fenced_pools.iter().map(usize::to_string).collect::>().join(", "); + Some(format!("skipped: bucket-volume heal fenced on decommission-suspended pool(s): {pools}")) +} + +async fn snapshot_heal_bucket_fence( + disks: &[Option], + pool_meta: Option<&RwLock>, + dispatch_fenced_pools: &[usize], +) -> Result<(Vec, BTreeSet)> { + let mut fenced_disks = vec![false; disks.len()]; + let mut fenced_pools = dispatch_fenced_pools.iter().copied().collect::>(); + let pool_meta = match pool_meta { + Some(pool_meta) => Some(pool_meta.read().await), + None => None, + }; + if let Some(pool_meta) = pool_meta.as_ref() + && let Some(pool_idx) = fenced_pools.iter().find(|pool_idx| **pool_idx >= pool_meta.pools.len()) + { + return Err(Error::other(format!( + "bucket-heal dispatch fence pool index {pool_idx} is absent from {} pool metadata entries", + pool_meta.pools.len() + ))); + } + + for (disk_index, disk) in disks.iter().enumerate() { + let Some(disk) = disk else { + continue; + }; + let pool_idx = disk_pool_index(disk)?; + if let Some(pool_meta) = pool_meta.as_ref() { + if pool_idx >= pool_meta.pools.len() { + return Err(Error::other(format!( + "bucket-heal pool index {pool_idx} is absent from {} pool metadata entries", + pool_meta.pools.len() + ))); + } + if pool_meta.is_suspended(pool_idx) { + fenced_pools.insert(pool_idx); + } + } + if fenced_pools.contains(&pool_idx) { + fenced_disks[disk_index] = true; + } + } + Ok((fenced_disks, fenced_pools)) +} + +async fn run_heal_bucket_volume_mutation( + disk: &DiskStore, + pool_meta: Option<&RwLock>, + operation: F, +) -> Result> +where + F: FnOnce() -> Fut, + Fut: std::future::Future>, +{ + let Some(pool_meta) = pool_meta else { + operation().await?; + return Ok(None); + }; + let pool_idx = disk_pool_index(disk)?; + let pool_meta = pool_meta.read().await; + if pool_idx >= pool_meta.pools.len() { + return Err(Error::other(format!( + "bucket-heal pool index {pool_idx} is absent from {} pool metadata entries", + pool_meta.pools.len() + ))); + } + if pool_meta.is_suspended(pool_idx) { + return Ok(Some(pool_idx)); + } + + // Keep the metadata read guard through the disk mutation so a decommission + // transition cannot pass between this state check and the destructive action. + operation().await?; + Ok(None) +} + +async fn heal_bucket_local_on_disks_with_pool_meta( + bucket: &str, + opts: &HealOpts, + disks: Vec>, + pool_meta: Option<&RwLock>, + dispatch_fenced_pools: &[usize], +) -> Result { + let (fenced_disks, mut fenced_pool_idxs) = snapshot_heal_bucket_fence(&disks, pool_meta, dispatch_fenced_pools).await?; + let fenced_disks = Arc::new(fenced_disks); let before_state = Arc::new(RwLock::new(vec![String::new(); disks.len()])); let after_state = Arc::new(RwLock::new(vec![String::new(); disks.len()])); @@ -1238,7 +1431,14 @@ pub(crate) async fn heal_bucket_local_on_disks( let bucket = bucket.to_string(); let bs_clone = before_state.clone(); let as_clone = after_state.clone(); + let fenced_disks = fenced_disks.clone(); futures.push(async move { + if fenced_disks[index] { + let skipped = fenced_decommission_drive_state().to_string(); + bs_clone.write().await[index] = skipped.clone(); + as_clone.write().await[index] = skipped; + return None; + } let disk = match disk { Some(disk) => disk, None => { @@ -1301,9 +1501,14 @@ pub(crate) async fn heal_bucket_local_on_disks( state: state.to_string(), }); } + if let Some(detail) = heal_bucket_fence_detail(&fenced_pool_idxs) { + res.detail = detail; + } return Ok(res); } + pause_before_heal_bucket_volume_mutation().await; + let mut operation_error = errs .iter() .filter_map(|err| match err { @@ -1315,26 +1520,35 @@ pub(crate) async fn heal_bucket_local_on_disks( if opts.remove && !bucket.starts_with(disk::RUSTFS_META_BUCKET) && !is_all_buckets_not_found(&errs) { let mut futures = Vec::new(); for (index, disk) in disks.iter().enumerate() { - if matches!(errs[index].as_ref(), Some(Error::DiskNotFound | Error::VolumeNotFound)) { + if fenced_disks[index] || matches!(errs[index].as_ref(), Some(Error::DiskNotFound | Error::VolumeNotFound)) { continue; } let Some(disk) = disk.clone() else { continue; }; let bucket = bucket.to_string(); + let mutation_disk = disk.clone(); futures.push(async move { - if let Some(err) = injected_heal_bucket_operation_error(&bucket, index, HealBucketOperation::Delete) { - return (index, Err(err)); - } - (index, disk.delete_volume(&bucket, false).await) + let result = run_heal_bucket_volume_mutation(&disk, pool_meta, || async move { + if let Some(err) = injected_heal_bucket_operation_error(&bucket, index, HealBucketOperation::Delete) { + return Err(err); + } + mutation_disk.delete_volume(&bucket, false).await + }) + .await; + (index, result) }); } for (index, result) in join_all(futures).await { match result { - Ok(()) | Err(Error::VolumeNotFound) => { + Ok(None) | Err(Error::VolumeNotFound) => { after_state.write().await[index] = DriveState::Missing.to_string(); } + Ok(Some(pool_idx)) => { + fenced_pool_idxs.insert(pool_idx); + after_state.write().await[index] = fenced_decommission_drive_state().to_string(); + } Err(Error::VolumeNotEmpty) => { warn!( bucket, @@ -1365,30 +1579,38 @@ pub(crate) async fn heal_bucket_local_on_disks( let bs_clone = before_state.clone(); futures.push(async move { if bs_clone.read().await[idx] == DriveState::Missing.to_string() { - let Some(disk) = disk.as_ref() else { - return (idx, Some(Error::DiskNotFound)); + let Some(disk) = disk else { + return (idx, Err(Error::DiskNotFound)); }; - - if let Some(err) = injected_heal_bucket_operation_error(&bucket, idx, HealBucketOperation::Make) { - return (idx, Some(err)); - } - match disk.make_volume(&bucket).await { - Ok(()) | Err(Error::VolumeExists) => return (idx, None), - Err(err) => return (idx, Some(err)), - } + let mutation_disk = disk.clone(); + let result = run_heal_bucket_volume_mutation(&disk, pool_meta, || async move { + if let Some(err) = injected_heal_bucket_operation_error(&bucket, idx, HealBucketOperation::Make) { + return Err(err); + } + match mutation_disk.make_volume(&bucket).await { + Ok(()) | Err(Error::VolumeExists) => Ok(()), + Err(err) => Err(err), + } + }) + .await; + return (idx, result); } - (idx, None) + (idx, Ok(None)) }); } for (index, result) in join_all(futures).await { match result { - None => { + Ok(None) => { if before_state.read().await[index] == DriveState::Missing.to_string() { after_state.write().await[index] = DriveState::Ok.to_string(); } } - Some(err) => { + Ok(Some(pool_idx)) => { + fenced_pool_idxs.insert(pool_idx); + after_state.write().await[index] = fenced_decommission_drive_state().to_string(); + } + Err(err) => { after_state.write().await[index] = match &err { Error::DiskNotFound => DriveState::Offline.to_string(), _ => DriveState::Corrupt.to_string(), @@ -1409,6 +1631,10 @@ pub(crate) async fn heal_bucket_local_on_disks( }); } + if let Some(detail) = heal_bucket_fence_detail(&fenced_pool_idxs) { + res.detail = detail; + } + match operation_error { Some(err) => Err(err), None => Ok(res), @@ -1426,6 +1652,7 @@ async fn clone_drives() -> Vec> { #[cfg(test)] mod tests { use super::*; + use crate::core::pools::{PoolDecommissionInfo, PoolStatus}; use crate::disk::WalkDirOptions; use crate::disk::disk_store::LocalDiskWrapper; use crate::disk::endpoint::Endpoint; @@ -1599,6 +1826,23 @@ mod tests { disks } + fn heal_bucket_pool_meta(suspended_pool: Option) -> PoolMeta { + PoolMeta { + pools: (0..2) + .map(|pool_idx| PoolStatus { + id: pool_idx, + cmd_line: format!("pool-{pool_idx}"), + last_update: ::time::OffsetDateTime::UNIX_EPOCH, + decommission: (suspended_pool == Some(pool_idx)).then(|| PoolDecommissionInfo { + start_time: Some(::time::OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }), + }) + .collect(), + ..Default::default() + } + } + fn test_remote_peer(addr: &str) -> RemotePeerS3Client { RemotePeerS3Client { pools: Some(vec![0]), @@ -1910,6 +2154,127 @@ mod tests { reset_local_disk_test_state().await; } + #[tokio::test] + #[serial] + async fn heal_bucket_rechecks_decommission_before_recreating_volume() { + reset_local_disk_test_state().await; + + let temp_dir = TempDir::new().expect("create temp dir for bucket-heal fence regression"); + let disks = init_test_local_disks_for_pools(&temp_dir, &[(0, 1), (1, 1)], "heal-bucket-mutation-fence").await; + let bucket = "fenced-recreate-bucket"; + disks[0] + .make_volume(bucket) + .await + .expect("active pool should start with the bucket volume"); + + let pool_meta = Arc::new(RwLock::new(heal_bucket_pool_meta(None))); + let barrier = install_heal_bucket_pre_mutation_barrier(); + let heal = tokio::spawn({ + let disks = disks.clone(); + let pool_meta = pool_meta.clone(); + async move { + heal_bucket_local_on_disks_with_pool_meta( + bucket, + &HealOpts { + recreate: true, + ..Default::default() + }, + disks.into_iter().map(Some).collect(), + Some(pool_meta.as_ref()), + &[], + ) + .await + } + }); + + barrier.wait_until_paused().await; + pool_meta.write().await.pools[1].decommission = Some(PoolDecommissionInfo { + start_time: Some(::time::OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }); + barrier.release(); + + let result = heal + .await + .expect("bucket-heal task should join") + .expect("suspended pool should be reported as skipped"); + assert!(result.detail.contains("skipped") && result.detail.contains('1')); + assert!(matches!(disks[1].stat_volume(bucket).await, Err(Error::VolumeNotFound))); + + reset_local_disk_test_state().await; + } + + #[tokio::test] + #[serial] + async fn heal_bucket_dispatch_fence_blocks_stale_active_peer_state() { + reset_local_disk_test_state().await; + + let temp_dir = TempDir::new().expect("create temp dir for stale bucket-heal peer regression"); + let disks = init_test_local_disks_for_pools(&temp_dir, &[(0, 1), (1, 1)], "heal-bucket-dispatch-fence").await; + let bucket = "dispatch-fenced-bucket"; + disks[0] + .make_volume(bucket) + .await + .expect("active pool should start with the bucket volume"); + let stale_pool_meta = RwLock::new(heal_bucket_pool_meta(None)); + + let result = heal_bucket_local_on_disks_with_pool_meta( + bucket, + &HealOpts { + recreate: true, + ..Default::default() + }, + disks.iter().cloned().map(Some).collect(), + Some(&stale_pool_meta), + &[1], + ) + .await + .expect("dispatch fence should override stale active peer metadata"); + + assert!(result.detail.contains("skipped") && result.detail.contains('1')); + assert!(matches!(disks[1].stat_volume(bucket).await, Err(Error::VolumeNotFound))); + + reset_local_disk_test_state().await; + } + + #[tokio::test] + #[serial] + async fn heal_bucket_keeps_suspended_pool_volume_on_remove() { + reset_local_disk_test_state().await; + + let temp_dir = TempDir::new().expect("create temp dir for bucket-heal delete fence regression"); + let disks = init_test_local_disks_for_pools(&temp_dir, &[(0, 1), (1, 1)], "heal-bucket-delete-fence").await; + let bucket = "fenced-remove-bucket"; + for disk in &disks { + disk.make_volume(bucket) + .await + .expect("bucket volume should exist before heal"); + } + let pool_meta = RwLock::new(heal_bucket_pool_meta(Some(1))); + + let result = heal_bucket_local_on_disks_with_pool_meta( + bucket, + &HealOpts { + remove: true, + ..Default::default() + }, + disks.iter().cloned().map(Some).collect(), + Some(&pool_meta), + &[], + ) + .await + .expect("suspended pool should be skipped during bucket-volume removal"); + + assert!(result.detail.contains("skipped") && result.detail.contains('1')); + assert!(matches!(disks[0].stat_volume(bucket).await, Err(Error::VolumeNotFound))); + disks[1] + .stat_volume(bucket) + .await + .expect("suspended pool bucket volume must not be deleted"); + + reset_local_disk_test_state().await; + } + #[tokio::test] #[serial] async fn heal_bucket_local_dry_run_reports_discovered_drive_states() { @@ -2123,6 +2488,32 @@ mod tests { assert!(partial.recreate); } + #[test] + fn heal_bucket_rpc_envelope_preserves_legacy_compatibility_fail_closed() { + let opts = HealOpts { + recreate: true, + pool: Some(2), + ..Default::default() + }; + let encoded = encode_heal_bucket_rpc_options(opts, &[1, 2]).expect("encode bucket-heal RPC envelope"); + + assert!( + serde_json::from_str::(&encoded).is_err(), + "an old peer must reject the nested request instead of ignoring its dispatch fence" + ); + let (decoded, fenced_pools) = + decode_heal_bucket_rpc_options(&encoded).expect("new peer should decode bucket-heal RPC envelope"); + assert!(decoded.recreate); + assert_eq!(decoded.pool, Some(2)); + assert_eq!(fenced_pools, vec![1, 2]); + + let legacy = serde_json::to_string(&opts).expect("encode legacy HealOpts"); + let (decoded, fenced_pools) = decode_heal_bucket_rpc_options(&legacy).expect("new peer should accept a legacy request"); + assert!(decoded.recreate); + assert_eq!(decoded.pool, Some(2)); + assert!(fenced_pools.is_empty()); + } + #[tokio::test] async fn test_make_bucket_reduces_quorum_by_pool_participants() { let peer_sys = S3PeerSys { diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index ffac77751..23dd3d537 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -14,6 +14,7 @@ use super::*; use crate::storage_api_contracts::heal::HealOperations as _; +use std::collections::BTreeSet; use tracing::trace; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; @@ -150,7 +151,45 @@ impl ECStore { #[instrument(skip(self))] pub(super) async fn handle_heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result { - let res = self.peer_sys.heal_bucket(bucket, opts).await?; + let mut fenced_pools = BTreeSet::new(); + { + let pool_meta = self.pool_meta.read().await; + fenced_pools.extend((0..pool_meta.pools.len()).filter(|pool_idx| pool_meta.is_suspended(*pool_idx))); + if let Some(pool_idx) = opts.pool { + if pool_idx >= pool_meta.pools.len() { + return Err(invalid_heal_pool_index(pool_idx, pool_meta.pools.len())); + } + if pool_meta.is_suspended(pool_idx) { + let complete = pool_meta.pools[pool_idx] + .decommission + .as_ref() + .is_some_and(|decommission| decommission.complete); + return Err(if complete { + StorageError::InvalidArgument( + "heal".to_string(), + "pool".to_string(), + format!("heal pool {pool_idx} has completed decommission"), + ) + } else { + Error::SlowDown + }); + } + } + } + + let dispatch_fenced_pools = fenced_pools.iter().copied().collect::>(); + let mut res = self + .peer_sys + .heal_bucket_with_fence(bucket, opts, &dispatch_fenced_pools) + .await?; + { + let pool_meta = self.pool_meta.read().await; + fenced_pools.extend((0..pool_meta.pools.len()).filter(|pool_idx| pool_meta.is_suspended(*pool_idx))); + } + if !fenced_pools.is_empty() { + let pools = fenced_pools.iter().map(usize::to_string).collect::>().join(", "); + res.detail = format!("skipped: bucket-volume heal fenced on decommission-suspended pool(s): {pools}"); + } Ok(res) } @@ -557,6 +596,64 @@ mod tests { } } + #[tokio::test] + async fn scoped_heal_bucket_blocks_before_dispatch_when_pool_is_suspended() { + let mut store = minimal_heal_store().await; + store.pool_meta = RwLock::new(PoolMeta { + pools: vec![ + PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: None, + }, + PoolStatus { + id: 1, + cmd_line: "pool-1".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }), + }, + ], + ..Default::default() + }); + + let err = store + .handle_heal_bucket( + "bucket", + &HealOpts { + pool: Some(1), + ..Default::default() + }, + ) + .await + .expect_err("suspended pool must be blocked before bucket-heal fan-out"); + assert_eq!(err, Error::SlowDown); + + store.pool_meta.write().await.pools[1] + .decommission + .as_mut() + .expect("decommission state should exist") + .complete = true; + let err = store + .handle_heal_bucket( + "bucket", + &HealOpts { + pool: Some(1), + ..Default::default() + }, + ) + .await + .expect_err("completed pool must remain fenced from bucket heal"); + assert!( + matches!(err, StorageError::InvalidArgument(_, ref field, ref reason) + if field == "pool" && reason.contains("completed decommission")), + "unexpected completed-pool error: {err:?}" + ); + } + #[tokio::test] #[serial_test::serial] async fn unscoped_heal_object_suspended_owner_semantics() { diff --git a/rustfs/src/storage/rpc/node_service/bucket.rs b/rustfs/src/storage/rpc/node_service/bucket.rs index 091fe028f..45e5ee248 100644 --- a/rustfs/src/storage/rpc/node_service/bucket.rs +++ b/rustfs/src/storage/rpc/node_service/bucket.rs @@ -17,9 +17,8 @@ use crate::storage::storage_api::rpc_consumer::node_service::contract::bucket::{ BucketOptions, DeleteBucketOptions, MakeBucketOptions, }; use crate::storage::storage_api::rpc_consumer::node_service::{ - DiskError, StoragePeerS3ClientExt as _, reload_bucket_metadata, remove_bucket_metadata, + DiskError, StoragePeerS3ClientExt as _, decode_heal_bucket_rpc_options, reload_bucket_metadata, remove_bucket_metadata, }; -use rustfs_common::heal_channel::HealOpts; use rustfs_protos::proto_gen::node_service::*; use tonic::{Request, Response, Status}; use tracing::debug; @@ -239,7 +238,7 @@ impl NodeService { ) -> Result, Status> { debug!("heal bucket"); let request = request.into_inner(); - let options = match serde_json::from_str::(&request.options) { + let (options, fenced_pools) = match decode_heal_bucket_rpc_options(&request.options) { Ok(options) => options, Err(err) => { return Ok(Response::new(HealBucketResponse { @@ -249,7 +248,11 @@ impl NodeService { } }; - match self.local_peer.heal_bucket(&request.bucket, &options).await { + match self + .local_peer + .heal_bucket_with_fence(&request.bucket, &options, &fenced_pools) + .await + { Ok(_) => Ok(Response::new(HealBucketResponse { success: true, error: None, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 512e255ee..2da707cb1 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -239,6 +239,7 @@ pub(crate) mod rpc_consumer { } pub(crate) mod node_service { + pub(crate) use super::super::ecstore_rpc::decode_heal_bucket_rpc_options; pub(crate) use super::super::storage_contracts::{ SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, }; @@ -520,10 +521,10 @@ pub(crate) mod ecstore_rpc { pub(crate) use rustfs_ecstore::api::rpc::{ KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, TONIC_RPC_PREFIX, - check_and_record_signed_rpc_nonce, normalize_tonic_rpc_audience, sign_ns_scanner_capability, sign_put_file_capability, - sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, - tonic_rpc_auth_failure_reason, verify_put_file_auth_trailer, verify_rpc_signature, verify_tonic_canonical_body_digest, - verify_tonic_mutation_body_digest, verify_tonic_rpc_signature_with_bootstrap, + check_and_record_signed_rpc_nonce, decode_heal_bucket_rpc_options, normalize_tonic_rpc_audience, + sign_ns_scanner_capability, sign_put_file_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, + tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, verify_put_file_auth_trailer, verify_rpc_signature, + verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_rpc_signature_with_bootstrap, }; #[cfg(test)] pub(crate) use rustfs_ecstore::api::rpc::{ @@ -1412,6 +1413,12 @@ pub(crate) trait StoragePeerS3ClientExt { bucket: &str, opts: &rustfs_common::heal_channel::HealOpts, ) -> DiskResult; + async fn heal_bucket_with_fence( + &self, + bucket: &str, + opts: &rustfs_common::heal_channel::HealOpts, + fenced_pools: &[usize], + ) -> DiskResult; async fn make_bucket(&self, bucket: &str, opts: &contract::bucket::MakeBucketOptions) -> DiskResult<()>; async fn list_bucket(&self, opts: &contract::bucket::BucketOptions) -> DiskResult>; async fn delete_bucket(&self, bucket: &str, opts: &contract::bucket::DeleteBucketOptions) -> DiskResult<()>; @@ -1431,6 +1438,15 @@ impl StoragePeerS3ClientExt for LocalPeerS3Client { ecstore_rpc::PeerS3Client::heal_bucket(self, bucket, opts).await } + async fn heal_bucket_with_fence( + &self, + bucket: &str, + opts: &rustfs_common::heal_channel::HealOpts, + fenced_pools: &[usize], + ) -> DiskResult { + ecstore_rpc::PeerS3Client::heal_bucket_with_fence(self, bucket, opts, fenced_pools).await + } + async fn make_bucket(&self, bucket: &str, opts: &contract::bucket::MakeBucketOptions) -> DiskResult<()> { ecstore_rpc::PeerS3Client::make_bucket(self, bucket, opts).await }