diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index 8910d896d..20e3cfef0 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -56,6 +56,7 @@ async fn peek_with_timeout(reader: &mut MetacacheReader pub(crate) enum TestReaderBehavior { Eof, Stall, + IgnoreCancel, ProducerError(DiskError), PartialThenTimeout(Vec), } @@ -72,6 +73,7 @@ pub struct ListPathRawOptions { pub min_disks: usize, pub report_not_found: bool, pub per_disk_limit: i32, + pub skip_walkdir_total_timeout: bool, pub agreed: Option, pub partial: Option, pub finished: Option, @@ -97,6 +99,7 @@ impl Clone for ListPathRawOptions { min_disks: self.min_disks, report_not_found: self.report_not_found, per_disk_limit: self.per_disk_limit, + skip_walkdir_total_timeout: self.skip_walkdir_total_timeout, #[cfg(test)] test_reader_behaviors: self.test_reader_behaviors.clone(), #[cfg(test)] @@ -137,6 +140,11 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d cancel_rx_clone.cancelled().await; return Ok(()); } + TestReaderBehavior::IgnoreCancel => { + let _held_writer = wr; + std::future::pending::<()>().await; + return Ok(()); + } TestReaderBehavior::ProducerError(err) => { record_producer_error(&producer_errs_clone, disk_idx, &err); return Err(err); @@ -162,6 +170,7 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d filter_prefix: opts_clone.filter_prefix.clone(), forward_to: opts_clone.forward_to.clone(), limit: opts_clone.per_disk_limit, + skip_total_timeout: opts_clone.skip_walkdir_total_timeout, ..Default::default() }; @@ -213,6 +222,7 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d filter_prefix: opts_clone.filter_prefix.clone(), forward_to: opts_clone.forward_to.clone(), limit: opts_clone.per_disk_limit, + skip_total_timeout: opts_clone.skip_walkdir_total_timeout, ..Default::default() }, &mut wr, @@ -485,6 +495,9 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d if let Err(err) = revjob.await.map_err(std::io::Error::other)? { error!("list_path_raw: revjob err {:?}", err); cancel_rx.cancel(); + for job in jobs { + job.abort(); + } return Err(err); } @@ -492,9 +505,15 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d // The merge consumer can finish successfully before every producer finishes // (for example after reaching EOF quorum while a tolerated drive is stalled, // or after the requested listing limit is satisfied). Cancel remaining walk - // jobs before joining them so list calls do not wait for slow remote streams. + // jobs before aborting them so list calls do not wait for slow remote streams. cancel_rx.cancel(); + for job in jobs.iter() { + if !job.is_finished() { + job.abort(); + } + } + let results = join_all(jobs).await; let mut job_errs = Vec::new(); for result in results { @@ -509,6 +528,9 @@ pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> d job_errs.push(err); } Err(err) => { + if err.is_cancelled() { + continue; + } error!("list_path_raw join err {:?}", err); job_errs.push(err.into()); } @@ -582,6 +604,31 @@ mod tests { .expect("listing should complete when healthy quorum reached EOF and only a tolerated drive stalled"); } + #[tokio::test] + async fn list_path_raw_aborts_unresponsive_producer_after_quorum_eof() { + let result = timeout( + Duration::from_millis(200), + list_path_raw( + CancellationToken::new(), + ListPathRawOptions { + disks: vec![None, None, None], + min_disks: 2, + test_reader_behaviors: vec![ + TestReaderBehavior::Eof, + TestReaderBehavior::Eof, + TestReaderBehavior::IgnoreCancel, + ], + peek_timeout: Some(Duration::from_millis(20)), + ..Default::default() + }, + ), + ) + .await; + + let listing = result.expect("list_path_raw should abort unresponsive producer instead of hanging"); + assert!(listing.is_ok()); + } + #[tokio::test] async fn list_path_raw_returns_timeout_when_producer_fails_after_partial_entry() { let seen = Arc::new(Mutex::new(Vec::new())); diff --git a/crates/ecstore/src/data_movement.rs b/crates/ecstore/src/data_movement.rs index 19342dbbb..3c1cb8b91 100644 --- a/crates/ecstore/src/data_movement.rs +++ b/crates/ecstore/src/data_movement.rs @@ -12,11 +12,14 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::error::{Error, Result}; +use crate::error::{Error, Result, is_err_data_movement_overwrite, is_err_object_not_found, is_err_version_not_found}; use crate::store::ECStore; -use crate::store_api::{CompletePart, GetObjectReader, MultipartOperations, ObjectIO, ObjectInfo, ObjectOptions, PutObjReader}; +use crate::store_api::{ + CompletePart, GetObjectReader, MultipartOperations, ObjectIO, ObjectInfo, ObjectOperations, ObjectOptions, PutObjReader, +}; use bytes::Bytes; use rustfs_rio::{EtagResolvable, HashReader, HashReaderDetector, Index, TryGetIndex}; +use rustfs_utils::path::encode_dir_object; use std::io::Cursor; use std::pin::Pin; use std::sync::{ @@ -102,13 +105,14 @@ fn data_movement_new_multipart_opts(object_info: &ObjectInfo, src_pool_idx: usiz } } -fn data_movement_complete_multipart_opts(object_info: &ObjectInfo) -> ObjectOptions { +fn data_movement_complete_multipart_opts(object_info: &ObjectInfo, src_pool_idx: usize) -> ObjectOptions { ObjectOptions { versioned: object_info.version_id.is_some(), version_id: object_info.version_id.as_ref().map(|v| v.to_string()), data_movement: true, mod_time: object_info.mod_time, preserve_etag: object_info.etag.clone(), + src_pool_idx, ..Default::default() } } @@ -139,6 +143,111 @@ fn resolve_data_movement_abort_result( )) } +fn data_movement_stage_error(op_label: &str, stage: &str, bucket: &str, object: &str, err: impl std::fmt::Display) -> Error { + Error::other(format!("{op_label}: {stage} failed for {bucket}/{object}: {err}")) +} + +fn should_check_data_movement_overwrite_resume(err: &Error) -> bool { + is_err_data_movement_overwrite(err) +} + +fn effective_actual_size(info: &ObjectInfo) -> Option { + info.get_actual_size().ok() +} + +fn is_equivalent_data_movement_object(source: &ObjectInfo, target: &ObjectInfo) -> bool { + source.version_id == target.version_id + && source.delete_marker == target.delete_marker + && source.size == target.size + && effective_actual_size(source) == effective_actual_size(target) + && source.etag == target.etag + && source.checksum == target.checksum + && source.mod_time == target.mod_time +} + +fn should_check_data_movement_resume_target(src_pool_idx: usize, target_pool_idx: usize) -> bool { + target_pool_idx != src_pool_idx +} + +async fn find_data_movement_target_info( + store: &ECStore, + target_pool_idx: usize, + bucket: &str, + object_info: &ObjectInfo, +) -> Result> { + let opts = ObjectOptions { + versioned: object_info.version_id.is_some(), + version_id: object_info.version_id.as_ref().map(|v| v.to_string()), + no_lock: true, + ..Default::default() + }; + let object = encode_dir_object(object_info.name.as_str()); + + let Some(pool) = store.pools.get(target_pool_idx) else { + return Err(Error::other(format!( + "data movement resume target pool {target_pool_idx} is out of range for {bucket}/{object}" + ))); + }; + + match pool.get_object_info(bucket, object.as_str(), &opts).await { + Ok(target_info) => Ok(Some(target_info)), + Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => Ok(None), + Err(err) => Err(err), + } +} + +fn resolve_data_movement_overwrite_resume_result( + err: &Error, + target_result: Result>, + source: &ObjectInfo, + src_pool_idx: usize, + target_pool_idx: usize, +) -> Result { + if !should_check_data_movement_overwrite_resume(err) + || !should_check_data_movement_resume_target(src_pool_idx, target_pool_idx) + { + return Ok(false); + } + + let Some(target) = target_result? else { + return Ok(false); + }; + + Ok(is_equivalent_data_movement_object(source, &target)) +} + +async fn should_treat_data_movement_overwrite_as_complete( + store: &ECStore, + src_pool_idx: usize, + target_pool_idx: usize, + bucket: &str, + object_info: &ObjectInfo, + err: &Error, +) -> Result { + if !should_check_data_movement_overwrite_resume(err) { + return Ok(false); + } + + resolve_data_movement_overwrite_resume_result( + err, + find_data_movement_target_info(store, target_pool_idx, bucket, object_info).await, + object_info, + src_pool_idx, + target_pool_idx, + ) +} + +fn data_movement_part_stage_error( + op_label: &str, + stage: &str, + bucket: &str, + object: &str, + part_number: usize, + err: impl std::fmt::Display, +) -> Error { + Error::other(format!("{op_label}: {stage} failed for {bucket}/{object} part {part_number}: {err}")) +} + pub(crate) async fn migrate_object( store: Arc, pool_idx: usize, @@ -149,14 +258,24 @@ pub(crate) async fn migrate_object( let object_info = rd.object_info.clone(); if object_info.is_multipart() { - let res = match store - .new_multipart_upload(&bucket, &object_info.name, &data_movement_new_multipart_opts(&object_info, pool_idx)) + let (res, target_pool_idx) = match store + .handle_new_multipart_upload_with_pool_idx( + &bucket, + &object_info.name, + &data_movement_new_multipart_opts(&object_info, pool_idx), + ) .await { Ok(res) => res, Err(err) => { error!("{op_label}: new_multipart_upload err {:?}", &err); - return Err(err); + return Err(data_movement_stage_error( + op_label, + "new_multipart_upload", + bucket.as_str(), + object_info.name.as_str(), + err, + )); } }; @@ -167,12 +286,39 @@ pub(crate) async fn migrate_object( for (i, part) in object_info.parts.iter().enumerate() { let mut chunk = vec![0u8; part.size]; - reader.read_exact(&mut chunk).await?; + reader.read_exact(&mut chunk).await.map_err(|err| { + data_movement_part_stage_error( + op_label, + "read_part", + bucket.as_str(), + object_info.name.as_str(), + part.number, + Error::other(err.to_string()), + ) + })?; - let part_size = i64::try_from(part.size).map_err(|_| Error::other("part size overflow"))?; + let part_size = i64::try_from(part.size).map_err(|_| { + data_movement_part_stage_error( + op_label, + "prepare_part", + bucket.as_str(), + object_info.name.as_str(), + part.number, + Error::other("part size overflow"), + ) + })?; let part_actual_size = if part.actual_size > 0 { part.actual_size } else { part_size }; let index = decode_part_index(part.index.as_ref()); - let mut data = put_obj_reader_from_chunk(chunk, part_size, part_actual_size, index)?; + let mut data = put_obj_reader_from_chunk(chunk, part_size, part_actual_size, index).map_err(|err| { + data_movement_part_stage_error( + op_label, + "prepare_part", + bucket.as_str(), + object_info.name.as_str(), + part.number, + err, + ) + })?; let pi = match store .put_object_part( @@ -191,7 +337,14 @@ pub(crate) async fn migrate_object( Ok(pi) => pi, Err(err) => { error!("{op_label}: put_object_part {i} err {:?}", &err); - return Err(err); + return Err(data_movement_part_stage_error( + op_label, + "put_object_part", + bucket.as_str(), + object_info.name.as_str(), + part.number, + err, + )); } }; @@ -209,12 +362,32 @@ pub(crate) async fn migrate_object( &object_info.name, &res.upload_id, parts, - &data_movement_complete_multipart_opts(&object_info), + &data_movement_complete_multipart_opts(&object_info, pool_idx), ) .await { + if should_treat_data_movement_overwrite_as_complete( + store.as_ref(), + pool_idx, + target_pool_idx, + bucket.as_str(), + &object_info, + &err, + ) + .await? + { + mark_multipart_upload_completed(&abort_multipart_flag); + return Ok(()); + } + error!("{op_label}: complete_multipart_upload err {:?}", &err); - return Err(err); + return Err(data_movement_stage_error( + op_label, + "complete_multipart_upload", + bucket.as_str(), + object_info.name.as_str(), + err, + )); } mark_multipart_upload_completed(&abort_multipart_flag); @@ -248,13 +421,18 @@ pub(crate) async fn migrate_object( return Ok(()); } - let actual_size = object_info.get_actual_size()?; + let actual_size = object_info.get_actual_size().map_err(|err| { + data_movement_stage_error(op_label, "prepare_put_object", bucket.as_str(), object_info.name.as_str(), err) + })?; let index = object_info .parts .first() .and_then(|part| decode_part_index(part.index.as_ref())); let reader = IndexedDataMovementReader::new(BufReader::new(rd.stream), index); - let hrd = HashReader::from_stream(reader, object_info.size, actual_size, object_info.etag.clone(), None, false)?; + let hrd = + HashReader::from_stream(reader, object_info.size, actual_size, object_info.etag.clone(), None, false).map_err(|err| { + data_movement_stage_error(op_label, "prepare_put_object", bucket.as_str(), object_info.name.as_str(), err) + })?; let mut data = PutObjReader::new(hrd); if let Err(err) = store @@ -267,7 +445,13 @@ pub(crate) async fn migrate_object( .await { error!("{op_label}: put_object err {:?}", &err); - return Err(err); + return Err(data_movement_stage_error( + op_label, + "put_object", + bucket.as_str(), + object_info.name.as_str(), + err, + )); } Ok(()) @@ -309,6 +493,33 @@ mod tests { assert!(message.contains(Error::SlowDown.to_string().as_str())); } + #[test] + fn test_data_movement_stage_error_includes_stage_and_object() { + let err = data_movement_stage_error("rebalance_object", "put_object", "bucket-a", "object-a", Error::SlowDown); + let message = err.to_string(); + assert!(message.contains("rebalance_object: put_object failed for bucket-a/object-a")); + assert!(message.contains(Error::SlowDown.to_string().as_str())); + } + + #[test] + fn test_data_movement_part_stage_error_includes_stage_object_and_part() { + let err = + data_movement_part_stage_error("rebalance_object", "put_object_part", "bucket-a", "object-a", 7, Error::SlowDown); + let message = err.to_string(); + assert!(message.contains("rebalance_object: put_object_part failed for bucket-a/object-a part 7")); + assert!(message.contains(Error::SlowDown.to_string().as_str())); + } + + #[test] + fn test_should_check_data_movement_overwrite_resume_only_for_overwrite_error() { + assert!(should_check_data_movement_overwrite_resume(&Error::DataMovementOverwriteErr( + "bucket-a".to_string(), + "object-a".to_string(), + "version-a".to_string(), + ))); + assert!(!should_check_data_movement_overwrite_resume(&Error::SlowDown)); + } + #[test] fn test_decode_part_index_returns_none_when_absent() { assert!(decode_part_index(None).is_none()); @@ -366,13 +577,14 @@ mod tests { ..Default::default() }; - let opts = data_movement_complete_multipart_opts(&object_info); + let opts = data_movement_complete_multipart_opts(&object_info, 7); assert!(opts.versioned); assert!(opts.data_movement); assert_eq!(opts.mod_time, Some(mod_time)); assert_eq!(opts.version_id.as_deref(), Some(version_id.to_string().as_str())); assert_eq!(opts.preserve_etag.as_deref(), Some("etag-value")); + assert_eq!(opts.src_pool_idx, 7); } #[test] @@ -396,4 +608,130 @@ mod tests { assert!(opts.data_movement); assert_eq!(opts.mod_time, object_info.mod_time); } + + #[test] + fn test_is_equivalent_data_movement_object_accepts_matching_metadata() { + let version_id = Uuid::nil(); + let info = ObjectInfo { + version_id: Some(version_id), + size: 128, + actual_size: 96, + etag: Some("etag-value".to_string()), + checksum: Some(Bytes::from_static(b"checksum")), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + + assert!(is_equivalent_data_movement_object(&info, &info.clone())); + } + + #[test] + fn test_is_equivalent_data_movement_object_rejects_content_mismatch() { + let source = ObjectInfo { + version_id: Some(Uuid::nil()), + size: 128, + actual_size: 96, + etag: Some("etag-source".to_string()), + checksum: Some(Bytes::from_static(b"checksum-source")), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + let target = ObjectInfo { + etag: Some("etag-target".to_string()), + checksum: Some(Bytes::from_static(b"checksum-target")), + ..source.clone() + }; + + assert!(!is_equivalent_data_movement_object(&source, &target)); + } + + #[test] + fn test_is_equivalent_data_movement_object_uses_effective_actual_size() { + let source = ObjectInfo { + size: 128, + actual_size: 0, + etag: Some("etag-value".to_string()), + ..Default::default() + }; + let target = ObjectInfo { + size: 128, + actual_size: 128, + etag: Some("etag-value".to_string()), + ..Default::default() + }; + + assert!(is_equivalent_data_movement_object(&source, &target)); + } + + #[test] + fn test_resolve_data_movement_overwrite_resume_result_accepts_equivalent_target() { + let source = ObjectInfo { + version_id: Some(Uuid::nil()), + size: 128, + etag: Some("etag-value".to_string()), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + let err = Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "version".to_string()); + + let should_resume = resolve_data_movement_overwrite_resume_result(&err, Ok(Some(source.clone())), &source, 0, 1) + .expect("equivalent overwrite target should be evaluated"); + + assert!(should_resume); + } + + #[test] + fn test_resolve_data_movement_overwrite_resume_result_rejects_source_pool_target() { + let source = ObjectInfo { + version_id: Some(Uuid::nil()), + size: 128, + etag: Some("etag-value".to_string()), + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + let err = Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "version".to_string()); + + let should_resume = resolve_data_movement_overwrite_resume_result(&err, Ok(Some(source.clone())), &source, 0, 0) + .expect("source-pool target should be rejected before target lookup"); + + assert!(!should_resume); + } + + #[test] + fn test_resolve_data_movement_overwrite_resume_result_rejects_non_equivalent_target() { + let source = ObjectInfo { + version_id: Some(Uuid::nil()), + size: 128, + etag: Some("etag-source".to_string()), + ..Default::default() + }; + let target = ObjectInfo { + etag: Some("etag-target".to_string()), + ..source.clone() + }; + let err = Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "version".to_string()); + + let should_resume = resolve_data_movement_overwrite_resume_result(&err, Ok(Some(target)), &source, 0, 1) + .expect("non-equivalent overwrite target should be evaluated"); + + assert!(!should_resume); + } + + #[test] + fn test_resolve_data_movement_overwrite_resume_result_propagates_target_lookup_error() { + let source = ObjectInfo::default(); + let err = Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "version".to_string()); + let result = resolve_data_movement_overwrite_resume_result(&err, Err(Error::SlowDown), &source, 0, 1); + + assert!(matches!(result, Err(Error::SlowDown))); + } + + #[test] + fn test_resolve_data_movement_overwrite_resume_result_ignores_non_overwrite_error() { + let source = ObjectInfo::default(); + let result = resolve_data_movement_overwrite_resume_result(&Error::SlowDown, Err(Error::FileAccessDenied), &source, 0, 1) + .expect("non-overwrite errors should not query target equivalence"); + + assert!(!result); + } } diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 5b3984c20..f7e6e2d4b 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -1136,10 +1136,16 @@ impl DiskAPI for LocalDiskWrapper { } async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> { + let timeout_duration = if opts.skip_total_timeout { + Duration::ZERO + } else { + get_drive_walkdir_timeout() + }; + self.track_disk_health_with_op_and_timeout_action( "walk_dir", || async { self.disk.walk_dir(opts, wr).await }, - get_drive_walkdir_timeout(), + timeout_duration, // Listing/scanner backpressure should fail only the current walk, not poison drive health. TimeoutHealthAction::IgnoreFailure, ) @@ -1647,6 +1653,52 @@ mod tests { .await; } + #[tokio::test] + async fn walk_dir_skip_total_timeout_keeps_stream_pending() { + temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("1"))], async { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = + Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8")).expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + let wrapper = LocalDiskWrapper::new(disk, false); + let bucket = "test-bucket"; + let object = "test-object"; + + wrapper.make_volume(bucket).await.expect("bucket should be created"); + + let mut file_info = FileInfo::new(&format!("{bucket}/{object}"), 1, 0); + file_info.volume = bucket.to_string(); + file_info.name = object.to_string(); + file_info.mod_time = Some(::time::OffsetDateTime::now_utc()); + file_info.erasure.index = 1; + + wrapper + .write_metadata("", bucket, object, file_info) + .await + .expect("object metadata should be written"); + + let mut writer = PendingWriter; + let result = tokio::time::timeout( + Duration::from_millis(20), + wrapper.walk_dir( + WalkDirOptions { + bucket: bucket.to_string(), + recursive: true, + skip_total_timeout: true, + ..Default::default() + }, + &mut writer, + ), + ) + .await; + + assert!(result.is_err(), "skip_total_timeout should leave backpressured walk pending"); + assert_eq!(wrapper.runtime_state(), RuntimeDriveHealthState::Online); + assert!(!wrapper.health.is_faulty()); + }) + .await; + } + #[tokio::test] async fn walk_dir_timeout_does_not_break_followup_stat_volume() { temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("1"))], async { diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 1ac60cd35..111885315 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -698,6 +698,10 @@ pub struct WalkDirOptions { // DiskID contains the disk ID of the disk. // Leave empty to not check disk ID. pub disk_id: String, + + // Skip the wrapper-level total timeout for long streaming walks. + #[serde(default)] + pub skip_total_timeout: bool, } #[derive(Clone, Debug, Default)] @@ -894,6 +898,7 @@ mod tests { forward_to: Some("object/path".to_string()), limit: 100, disk_id: "disk-123".to_string(), + skip_total_timeout: false, }; assert_eq!(opts.bucket, "test-bucket"); @@ -904,6 +909,7 @@ mod tests { assert_eq!(opts.forward_to, Some("object/path".to_string())); assert_eq!(opts.limit, 100); assert_eq!(opts.disk_id, "disk-123"); + assert!(!opts.skip_total_timeout); } /// Test DeleteOptions structure diff --git a/crates/ecstore/src/rebalance.rs b/crates/ecstore/src/rebalance.rs index 6a19d5a4c..7463fa0be 100644 --- a/crates/ecstore/src/rebalance.rs +++ b/crates/ecstore/src/rebalance.rs @@ -19,18 +19,18 @@ use crate::data_movement; use crate::data_usage::DATA_USAGE_CACHE_NAME; use crate::disk::error::DiskError; use crate::error::{Error, Result}; -use crate::error::{ - is_err_data_movement_overwrite, is_err_object_not_found, is_err_operation_canceled, is_err_version_not_found, -}; +use crate::error::{is_err_object_not_found, is_err_operation_canceled, is_err_version_not_found, is_network_or_host_down}; use crate::global::get_global_endpoints; use crate::pools::ListCallback; -use crate::set_disk::SetDisks; +use crate::set_disk::{SetDisks, get_lock_acquire_timeout}; use crate::store::ECStore; use crate::store_api::{GetObjectReader, HTTPRangeSpec, ObjectIO, ObjectInfo, ObjectOperations, ObjectOptions}; use http::HeaderMap; +use rand::RngExt as _; use rustfs_filemeta::{FileInfo, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams}; use rustfs_utils::path::encode_dir_object; use serde::{Deserialize, Serialize}; +use std::collections::HashSet; use std::fmt; use std::future::Future; use std::io::Cursor; @@ -38,12 +38,17 @@ use std::sync::Arc; use time::OffsetDateTime; use tokio::time::{Duration, Instant}; use tokio_util::sync::CancellationToken; -use tracing::{error, info}; +use tracing::{error, info, warn}; use uuid::Uuid; const REBAL_META_FMT: u16 = 1; // Replace with actual format value const REBAL_META_VER: u16 = 1; // Replace with actual version value const REBAL_META_NAME: &str = "rebalance.bin"; +const REBALANCE_LISTING_MAX_ATTEMPTS: usize = 3; +const REBALANCE_LISTING_RETRY_BASE_DELAY: Duration = Duration::from_millis(250); +const REBALANCE_MIGRATION_RETRY_BASE_DELAY: Duration = Duration::from_millis(250); +const REBALANCE_MIGRATION_LOCK_RETRY_CAP: Duration = Duration::from_secs(10); +const REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX: &str = "deferred transient rebalance entry failure:"; #[derive(Debug, Default, Clone, Serialize, Deserialize)] pub struct RebalanceStats { @@ -93,6 +98,12 @@ impl RebalanceStats { self.bucket = bucket; self.object = fi.name.clone(); } + + pub fn update_batch(&mut self, bucket: String, versions: &[&FileInfo]) { + for version in versions { + self.update(bucket.clone(), version); + } + } } pub type RStats = Vec>; @@ -110,9 +121,22 @@ pub(crate) struct MigrationVersionResult { pub ignored: bool, pub cleanup_ignored: bool, pub failed: bool, + pub stage: Option<&'static str>, pub error: Option, } +#[derive(Debug, Clone, PartialEq, Eq)] +enum RebalanceBucketOutcome { + Completed, + Deferred { last_error: String }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum RebalanceEntryOutcome { + Completed, + Deferred { last_error: String }, +} + fn rebalance_delete_marker_opts(version: &FileInfo, version_id: Option, src_pool_idx: usize) -> ObjectOptions { ObjectOptions { versioned: true, @@ -198,12 +222,45 @@ pub(crate) async fn migrate_entry_version( version_id: Option, max_attempts: usize, ignore_data_usage_cache: bool, - mut transfer: F, + transfer: F, ) -> MigrationVersionResult where Backend: MigrationBackend + ?Sized, F: FnMut(usize, String, GetObjectReader) -> Fut + Send, Fut: Future> + Send, +{ + migrate_entry_version_with_retry_wait( + set, + bucket, + pool_index, + version, + version_id, + max_attempts, + ignore_data_usage_cache, + transfer, + sleep_rebalance_migration_retry, + ) + .await +} + +#[allow(clippy::too_many_arguments)] +async fn migrate_entry_version_with_retry_wait( + set: &Backend, + bucket: String, + pool_index: usize, + version: &FileInfo, + version_id: Option, + max_attempts: usize, + ignore_data_usage_cache: bool, + mut transfer: F, + mut wait_retry: W, +) -> MigrationVersionResult +where + Backend: MigrationBackend + ?Sized, + F: FnMut(usize, String, GetObjectReader) -> Fut + Send, + Fut: Future> + Send, + W: FnMut(Duration) -> WFut + Send, + WFut: Future + Send, { let max_attempts = max_attempts.max(1); @@ -213,6 +270,7 @@ where ignored: true, cleanup_ignored: false, failed: false, + stage: None, error: None, }; } @@ -227,12 +285,13 @@ where ) .await { - if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) { + if is_err_object_not_found(&err) || is_err_version_not_found(&err) { return MigrationVersionResult { moved: false, ignored: true, cleanup_ignored: true, failed: false, + stage: Some("move_remote_version"), error: None, }; } @@ -242,6 +301,7 @@ where ignored: false, cleanup_ignored: false, failed: true, + stage: Some("move_remote_version"), error: Some(err), }; } @@ -251,6 +311,7 @@ where ignored: false, cleanup_ignored: false, failed: false, + stage: None, error: None, }; } @@ -260,12 +321,13 @@ where .delete_object_for_migration(&bucket, &version.name, rebalance_delete_marker_opts(version, version_id, pool_index)) .await { - if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) { + if is_err_object_not_found(&err) || is_err_version_not_found(&err) { return MigrationVersionResult { moved: false, ignored: true, cleanup_ignored: true, failed: false, + stage: Some("delete_marker"), error: None, }; } @@ -275,6 +337,7 @@ where ignored: false, cleanup_ignored: false, failed: true, + stage: Some("delete_marker"), error: Some(err), }; } @@ -284,6 +347,7 @@ where ignored: false, cleanup_ignored: false, failed: false, + stage: None, error: None, }; } @@ -312,47 +376,59 @@ where ignored: true, cleanup_ignored: true, failed: false, + stage: Some("read_source"), error: None, }; } last_error = Some(err); - if attempt + 1 >= max_attempts { + let Some(err) = last_error.as_ref() else { + continue; + }; + if attempt + 1 >= max_attempts || !is_transient_rebalance_error(err) { return MigrationVersionResult { moved: false, ignored: false, cleanup_ignored: false, failed: true, + stage: Some("read_source"), error: last_error, }; } + wait_retry(rebalance_migration_retry_delay(attempt, err)).await; continue; } }; if let Err(err) = transfer(pool_index, bucket.clone(), rd).await { - if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) { + if is_err_object_not_found(&err) || is_err_version_not_found(&err) { return MigrationVersionResult { moved: false, ignored: true, cleanup_ignored: true, failed: false, + stage: Some("write_target"), error: None, }; } last_error = Some(err); - if attempt + 1 >= max_attempts { + let Some(err) = last_error.as_ref() else { + continue; + }; + if attempt + 1 >= max_attempts || !is_transient_rebalance_error(err) { return MigrationVersionResult { moved: false, ignored: false, cleanup_ignored: false, failed: true, + stage: Some("write_target"), error: last_error, }; } + wait_retry(rebalance_migration_retry_delay(attempt, err)).await; continue; } @@ -361,6 +437,7 @@ where ignored: false, cleanup_ignored: false, failed: false, + stage: None, error: None, }; } @@ -370,6 +447,7 @@ where ignored: false, cleanup_ignored: false, failed: true, + stage: Some("migrate"), error: last_error, } } @@ -546,6 +624,36 @@ impl RebalanceMeta { } impl ECStore { + async fn save_rebalance_meta_with_merge( + &self, + pool: Arc, + local_snapshot: &RebalanceMeta, + stage: &str, + ) -> Result<()> { + let ns_lock = pool.new_ns_lock(crate::disk::RUSTFS_META_BUCKET, REBAL_META_NAME).await?; + let _guard = ns_lock + .get_write_lock(get_lock_acquire_timeout()) + .await + .map_err(rebalance_meta_lock_error)?; + + let opts = ObjectOptions { + no_lock: true, + ..Default::default() + }; + let mut merged = RebalanceMeta::new(); + match merged.load_with_opts(pool.clone(), opts.clone()).await { + Ok(()) => { + merge_rebalance_meta(&mut merged, local_snapshot); + } + Err(Error::ConfigNotFound) => { + merged = local_snapshot.clone(); + } + Err(err) => return Err(Error::other(format!("rebalance meta load before save failed during {stage}: {err}"))), + } + + merged.save_with_opts(pool, opts).await + } + #[tracing::instrument(skip_all)] pub async fn load_rebalance_meta(&self) -> Result<()> { let mut meta = RebalanceMeta::new(); @@ -602,7 +710,11 @@ impl ECStore { let rebalance_meta = self.rebalance_meta.read().await; if let Some(meta) = rebalance_meta.as_ref() { let pool = clone_first_arc(&self.pools, "update_rebalance_stats: no pools available")?; - resolve_rebalance_meta_save_result(meta.save(pool).await, "update_rebalance_stats")?; + resolve_rebalance_meta_save_result( + self.save_rebalance_meta_with_merge(pool, meta, "update_rebalance_stats") + .await, + "update_rebalance_stats", + )?; } } @@ -673,7 +785,10 @@ impl ECStore { }; let pool = clone_first_arc(&self.pools, "init_rebalance_meta: no pools available")?; - resolve_rebalance_meta_save_result(meta.save(pool).await, "init_rebalance_meta")?; + resolve_rebalance_meta_save_result( + self.save_rebalance_meta_with_merge(pool, &meta, "init_rebalance_meta").await, + "init_rebalance_meta", + )?; info!("init_rebalance_meta: rebalance meta saved"); @@ -690,11 +805,24 @@ impl ECStore { #[tracing::instrument(skip(self, fi))] pub async fn update_pool_stats(&self, pool_index: usize, bucket: String, fi: &FileInfo) -> Result<()> { + self.update_pool_stats_batch(pool_index, bucket, &[fi]).await + } + + #[tracing::instrument(skip(self, versions))] + pub async fn update_pool_stats_batch(&self, pool_index: usize, bucket: String, versions: &[&FileInfo]) -> Result<()> { + if versions.is_empty() { + return Ok(()); + } + let mut rebalance_meta = self.rebalance_meta.write().await; - if let Some(meta) = rebalance_meta.as_mut() - && let Some(pool_stat) = meta.pool_stats.get_mut(pool_index) - { - pool_stat.update(bucket, fi); + if let Some(meta) = rebalance_meta.as_mut() { + if !should_accept_rebalance_stats_update(meta, pool_index) { + return Ok(()); + } + + if let Some(pool_stat) = meta.pool_stats.get_mut(pool_index) { + pool_stat.update_batch(bucket, versions); + } } Ok(()) @@ -714,6 +842,23 @@ impl ECStore { mark_rebalance_bucket_done(rebalance_meta.as_mut(), pool_index, &bucket) } + async fn defer_rebalance_bucket(&self, pool_index: usize, bucket: String, last_error: String) -> Result<()> { + let mut rebalance_meta = self.rebalance_meta.write().await; + let Some(meta) = rebalance_meta.as_mut() else { + return Err(rebalance_metadata_not_initialized_error("defer rebalance bucket")); + }; + let pool_count = meta.pool_stats.len(); + ensure_valid_rebalance_pool_index(pool_count, pool_index)?; + let Some(pool_stat) = meta.pool_stats.get_mut(pool_index) else { + return Err(invalid_rebalance_pool_index_error(pool_index, pool_count)); + }; + + defer_bucket_in_rebalance_queue(pool_stat, &bucket)?; + pool_stat.info.last_error = Some(last_error); + meta.last_refreshed_at = Some(OffsetDateTime::now_utc()); + Ok(()) + } + pub async fn is_rebalance_started(&self) -> bool { let rebalance_meta = self.rebalance_meta.read().await; if let Some(meta) = rebalance_meta.as_ref() { @@ -766,7 +911,11 @@ impl ECStore { if let Some(meta_to_save) = meta_to_save { let pool = clone_first_arc(self.pools.as_slice(), "stop_rebalance: no pools available")?; - resolve_rebalance_meta_save_result(meta_to_save.save(pool).await, "stop_rebalance")?; + resolve_rebalance_meta_save_result( + self.save_rebalance_meta_with_merge(pool, &meta_to_save, "stop_rebalance") + .await, + "stop_rebalance", + )?; } Ok(()) @@ -803,7 +952,11 @@ impl ECStore { if let Some(meta) = meta_to_save { let pool = clone_first_arc(self.pools.as_slice(), "start_rebalance: no pools available")?; - resolve_rebalance_meta_save_result(meta.save(pool).await, "start_rebalance complete pools at goal")?; + resolve_rebalance_meta_save_result( + self.save_rebalance_meta_with_merge(pool, &meta, "start_rebalance complete pools at goal") + .await, + "start_rebalance complete pools at goal", + )?; } let participants = if let Some(ref meta) = *self.rebalance_meta.read().await { @@ -918,6 +1071,7 @@ impl ECStore { info!("Pool {} rebalancing is started", pool_index); let mut final_result: Result<()> = Ok(()); + let mut deferred_buckets = HashSet::new(); loop { if rx.is_cancelled() { @@ -945,17 +1099,45 @@ impl ECStore { if let Some(bucket) = next_bucket { info!("Rebalancing bucket: start {}", bucket); - if let Err(err) = resolve_rebalance_bucket_result( + let outcome = match resolve_rebalance_bucket_result( self.rebalance_bucket(rx.clone(), bucket.clone(), pool_index).await, pool_index, &bucket, ) { - error!("Error rebalancing bucket {}: {:?}", bucket, err); - final_result = Err(resolve_rebalance_terminal_error( - err.clone(), - send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await, - )); - break; + Ok(outcome) => outcome, + Err(err) => { + error!("Error rebalancing bucket {}: {:?}", bucket, err); + final_result = Err(resolve_rebalance_terminal_error( + err.clone(), + send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await, + )); + break; + } + }; + + if let RebalanceBucketOutcome::Deferred { last_error } = outcome { + if !deferred_buckets.insert(bucket.clone()) { + let err = Error::other(format!( + "rebalance bucket {bucket} deferred repeatedly due to transient object failures: {last_error}" + )); + error!("Error rebalancing bucket {}: {:?}", bucket, err); + final_result = Err(resolve_rebalance_terminal_error( + err.clone(), + send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await, + )); + break; + } + + warn!("Rebalance bucket deferred due to transient object failures: {bucket}: {last_error}"); + if let Err(err) = self.defer_rebalance_bucket(pool_index, bucket.clone(), last_error).await { + error!("defer_rebalance_bucket failed for pool {}: {:?}", pool_index, err); + final_result = Err(resolve_rebalance_terminal_error( + err.clone(), + send_rebalance_done_signal(&done_tx, Err(err.clone()), pool_index).await, + )); + break; + } + continue; } info!("Rebalance bucket: done {} ", bucket); @@ -1009,12 +1191,14 @@ impl ECStore { (pool_stat.init_free_space + pool_stat.bytes) as f64 / pool_stat.init_capacity as f64 }; - if rebalance_goal_reached( - pool_stat.init_free_space, - pool_stat.init_capacity, - pool_stat.bytes, - meta.percent_free_goal, - ) { + if !has_deferred_rebalance_error(pool_stat) + && rebalance_goal_reached( + pool_stat.init_free_space, + pool_stat.init_capacity, + pool_stat.bytes, + meta.percent_free_goal, + ) + { pool_stat.info.status = RebalStatus::Completed; pool_stat.info.end_time = Some(OffsetDateTime::now_utc()); info!("check_if_rebalance_done: pool {} is completed, pfi: {}", pool_index, pfi); @@ -1065,6 +1249,16 @@ fn clone_rebalance_pool_stats(meta: Option<&RebalanceMeta>) -> Result bool { + if meta.stopped_at.is_some() { + return false; + } + + meta.pool_stats + .get(pool_index) + .is_some_and(|pool_stat| pool_stat.info.status == RebalStatus::Started) +} + fn resolve_next_rebalance_bucket(meta: Option<&RebalanceMeta>, pool_index: usize) -> Result> { let Some(meta) = meta else { return Err(rebalance_metadata_not_initialized_error("resolve next rebalance bucket")); @@ -1108,6 +1302,9 @@ fn mark_rebalance_bucket_done(meta: Option<&mut RebalanceMeta>, pool_index: usiz if take_bucket_from_rebalance_queue(pool_stat, bucket) { info!("bucket_rebalance_done: bucket {} rebalanced", bucket); + if has_deferred_rebalance_error(pool_stat) { + pool_stat.info.last_error = None; + } Ok(()) } else { Err(Error::other(format!( @@ -1131,6 +1328,16 @@ fn take_bucket_from_rebalance_queue(pool_stat: &mut RebalanceStats, bucket: &str found } +fn defer_bucket_in_rebalance_queue(pool_stat: &mut RebalanceStats, bucket: &str) -> Result<()> { + let Some(pos) = pool_stat.buckets.iter().position(|name| name == bucket) else { + return Err(Error::other(format!("failed to defer rebalance bucket {bucket}: bucket was not queued"))); + }; + + let bucket = pool_stat.buckets.remove(pos); + pool_stat.buckets.push(bucket); + Ok(()) +} + fn should_pool_participate(init_free_space: u64, init_capacity: u64, percent_free_goal: f64) -> bool { init_capacity > 0 && percent_free_ratio(init_free_space, init_capacity) < percent_free_goal } @@ -1139,7 +1346,7 @@ fn complete_rebalance_pools_at_goal(meta: &mut RebalanceMeta, now: OffsetDateTim let mut changed = false; for pool_stat in meta.pool_stats.iter_mut() { - if !is_rebalance_pool_started(pool_stat) { + if !is_rebalance_pool_started(pool_stat) || has_deferred_rebalance_error(pool_stat) { continue; } @@ -1159,28 +1366,45 @@ fn complete_rebalance_pools_at_goal(meta: &mut RebalanceMeta, now: OffsetDateTim changed } -fn resolve_rebalance_worker_result( +fn has_deferred_rebalance_error(pool_stat: &RebalanceStats) -> bool { + pool_stat + .info + .last_error + .as_deref() + .is_some_and(|last_error| last_error.starts_with(REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX)) +} + +fn resolve_rebalance_worker_result( set_idx: usize, - worker_result: std::result::Result, tokio::task::JoinError>, -) -> Result<()> { + worker_result: std::result::Result, tokio::task::JoinError>, +) -> Result { match worker_result { Ok(result) => result, Err(err) => Err(Error::other(format!("rebalance worker {set_idx} task join error: {err}"))), } } -type RebalanceEntryTask = tokio::task::JoinHandle>; +type RebalanceEntryTask = tokio::task::JoinHandle>; -async fn wait_rebalance_entry_tasks(set_idx: usize, tasks: Arc>>) -> Result<()> { +async fn wait_rebalance_entry_tasks( + set_idx: usize, + tasks: Arc>>, +) -> Result> { let tasks = { let mut tasks = tasks.lock().await; std::mem::take(&mut *tasks) }; let mut first_error = None; + let mut first_deferred = None; for task in tasks { match task.await { - Ok(Ok(())) => {} + Ok(Ok(RebalanceEntryOutcome::Completed)) => {} + Ok(Ok(RebalanceEntryOutcome::Deferred { last_error })) => { + if first_deferred.is_none() { + first_deferred = Some(last_error); + } + } Ok(Err(err)) => { error!("rebalance entry task failed for set {}: {}", set_idx, err); if first_error.is_none() { @@ -1197,7 +1421,11 @@ async fn wait_rebalance_entry_tasks(set_idx: usize, tasks: Arc, stage: &str) -> Result result.map_err(|err| Error::other(format!("rebalance meta save failed during {stage}: {err}"))) } +fn rebalance_meta_lock_error(err: rustfs_lock::LockError) -> Error { + match err { + rustfs_lock::LockError::QuorumNotReached { required, achieved } => Error::NamespaceLockQuorumUnavailable { + mode: "write", + bucket: crate::disk::RUSTFS_META_BUCKET.to_string(), + object: REBAL_META_NAME.to_string(), + required, + achieved, + }, + other => Error::other(format!( + "failed to acquire rebalance metadata write lock on {}/{}: {other}", + crate::disk::RUSTFS_META_BUCKET, + REBAL_META_NAME + )), + } +} + fn resolve_rebalance_meta_load_result(result: Result<()>) -> Result { match result { Ok(()) => Ok(true), @@ -1267,6 +1512,10 @@ fn resolve_rebalance_migrate_result_error( }) } +fn should_defer_rebalance_entry_failure(err: &Error) -> bool { + is_transient_rebalance_error(err) +} + fn resolve_load_rebalance_stats_update_result(result: Result<()>) -> Result<()> { result.map_err(|err| Error::other(format!("rebalance metadata stats refresh failed after load: {err}"))) } @@ -1301,14 +1550,120 @@ fn resolve_rebalance_bucket_error(entry_error: Option, worker_error: Opti Ok(()) } -fn resolve_rebalance_bucket_result(result: Result<()>, pool_idx: usize, bucket: &str) -> Result<()> { +fn resolve_rebalance_bucket_result( + result: Result, + pool_idx: usize, + bucket: &str, +) -> Result { match result { - Ok(()) => Ok(()), + Ok(outcome) => Ok(outcome), Err(err) if is_err_operation_canceled(&err) => Err(err), Err(err) => Err(Error::other(format!("rebalance bucket {bucket} failed for pool {pool_idx}: {err}"))), } } +fn is_transient_rebalance_error(err: &Error) -> bool { + match err { + Error::SlowDown + | Error::ErasureReadQuorum + | Error::ErasureWriteQuorum + | Error::InsufficientReadQuorum(_, _) + | Error::InsufficientWriteQuorum(_, _) => true, + Error::Lock(lock_err) => is_rebalance_transient_lock_error(lock_err), + Error::Io(io_err) => is_rebalance_transient_io_error(io_err) || is_rebalance_transient_message(&io_err.to_string()), + _ => is_rebalance_transient_message(&err.to_string()) || is_network_or_host_down(&err.to_string(), true), + } +} + +fn is_rebalance_transient_lock_error(err: &rustfs_lock::LockError) -> bool { + match err { + rustfs_lock::LockError::Timeout { .. } | rustfs_lock::LockError::Network { .. } => true, + rustfs_lock::LockError::Internal { message } => is_rebalance_transient_message(message), + _ => false, + } +} + +fn is_rebalance_transient_io_error(err: &std::io::Error) -> bool { + if err.kind() == std::io::ErrorKind::TimedOut { + return true; + } + + if let Some(disk_err) = err.get_ref().and_then(|err| err.downcast_ref::()) + && *disk_err == DiskError::Timeout + { + return true; + } + + let message = err.to_string(); + message.eq_ignore_ascii_case("timeout") || is_rebalance_transient_message(&message) +} + +fn is_rebalance_transient_message(message: &str) -> bool { + let message = message.to_ascii_lowercase(); + message.contains("lock acquisition timed out") + || message.contains("remote lock rpc timed out") + || message.contains("keepalivetimedout") + || message.contains("i/o timeout") + || message.contains("operation timed out") +} + +fn should_retry_rebalance_listing(err: &Error, attempt: usize, max_attempts: usize) -> bool { + attempt + 1 < max_attempts && is_transient_rebalance_error(err) +} + +fn rebalance_listing_retry_delay(attempt: usize) -> Duration { + let multiplier = u32::try_from(attempt.saturating_add(1)).unwrap_or(u32::MAX); + REBALANCE_LISTING_RETRY_BASE_DELAY.saturating_mul(multiplier) +} + +fn is_rebalance_lock_or_rpc_timeout(err: &Error) -> bool { + match err { + Error::Lock(rustfs_lock::LockError::Timeout { .. }) | Error::Lock(rustfs_lock::LockError::Network { .. }) => true, + Error::Io(io_err) => is_rebalance_lock_or_rpc_timeout_message(&io_err.to_string()), + _ => is_rebalance_lock_or_rpc_timeout_message(&err.to_string()), + } +} + +fn is_rebalance_lock_or_rpc_timeout_message(message: &str) -> bool { + let message = message.to_ascii_lowercase(); + message.contains("lock acquisition timed out") + || message.contains("remote lock rpc timed out") + || message.contains("keepalivetimedout") +} + +fn rebalance_migration_retry_delay(attempt: usize, err: &Error) -> Duration { + if is_rebalance_lock_or_rpc_timeout(err) { + return rebalance_lock_retry_delay(attempt); + } + + let multiplier = u32::try_from(attempt.saturating_add(1)).unwrap_or(u32::MAX); + REBALANCE_MIGRATION_RETRY_BASE_DELAY.saturating_mul(multiplier) +} + +fn rebalance_lock_retry_delay(attempt: usize) -> Duration { + let lock_timeout = get_lock_acquire_timeout(); + let attempt_shift = u32::try_from(attempt.min(4)).unwrap_or(4); + let multiplier = 1_u32.checked_shl(attempt_shift).unwrap_or(u32::MAX); + let cap = lock_timeout + .saturating_mul(multiplier) + .min(REBALANCE_MIGRATION_LOCK_RETRY_CAP) + .max(REBALANCE_MIGRATION_RETRY_BASE_DELAY); + let max_millis = u64::try_from(cap.as_millis()).unwrap_or(u64::MAX).max(1); + let jitter_millis = rand::rng().random_range(1..=max_millis); + Duration::from_millis(jitter_millis) +} + +async fn sleep_rebalance_migration_retry(delay: Duration) { + tokio::time::sleep(delay).await; +} + +async fn wait_rebalance_listing_retry(rx: &CancellationToken, delay: Duration) -> Result<()> { + tokio::select! { + _ = rx.cancelled() => Err(Error::OperationCanceled), + _ = tokio::time::sleep(delay) => Ok(()), + } +} + fn ensure_rebalance_listing_disks_available(has_disks: bool, bucket: &str) -> Result<()> { if !has_disks { return Err(Error::other(format!( @@ -1523,6 +1878,113 @@ fn apply_rebalance_save_option(meta: &mut RebalanceMeta, pool_idx: usize, opt: R meta.last_refreshed_at = Some(now); } +fn is_rebalance_terminal_status(status: RebalStatus) -> bool { + matches!(status, RebalStatus::Completed | RebalStatus::Stopped | RebalStatus::Failed) +} + +fn merge_rebalance_bucket_lists(remote: &mut Vec, local: &[String]) { + let mut existing: HashSet = remote.iter().cloned().collect(); + for bucket in local { + if existing.insert(bucket.clone()) { + remote.push(bucket.clone()); + } + } +} + +fn remove_rebalanced_buckets_from_queue(pool_stat: &mut RebalanceStats) { + let rebalanced_buckets: HashSet = pool_stat.rebalanced_buckets.iter().cloned().collect(); + pool_stat.buckets.retain(|bucket| !rebalanced_buckets.contains(bucket)); +} + +fn merge_rebalance_pool_stats(remote: &mut RebalanceStats, local: &RebalanceStats) { + remote.init_free_space = remote.init_free_space.max(local.init_free_space); + remote.init_capacity = remote.init_capacity.max(local.init_capacity); + remote.participating |= local.participating; + + merge_rebalance_bucket_lists(&mut remote.buckets, &local.buckets); + merge_rebalance_bucket_lists(&mut remote.rebalanced_buckets, &local.rebalanced_buckets); + remove_rebalanced_buckets_from_queue(remote); + + let local_is_newer = local.num_versions >= remote.num_versions; + remote.num_objects = remote.num_objects.max(local.num_objects); + remote.num_versions = remote.num_versions.max(local.num_versions); + remote.bytes = remote.bytes.max(local.bytes); + + if local_is_newer { + remote.bucket = local.bucket.clone(); + remote.object = local.object.clone(); + } + + if remote.info.start_time.is_none() { + remote.info.start_time = local.info.start_time; + } + + if is_rebalance_terminal_status(remote.info.status) && local.info.status == RebalStatus::Started { + return; + } + + if remote.info.status == RebalStatus::Stopped && matches!(local.info.status, RebalStatus::Started | RebalStatus::Completed) { + return; + } + + match local.info.status { + RebalStatus::Failed => { + remote.info.status = RebalStatus::Failed; + remote.info.end_time = local.info.end_time.or(remote.info.end_time); + remote.info.last_error = local.info.last_error.clone().or_else(|| remote.info.last_error.clone()); + } + RebalStatus::Stopped => { + if remote.info.status != RebalStatus::Failed { + remote.info.status = RebalStatus::Stopped; + remote.info.end_time = local.info.end_time.or(remote.info.end_time); + remote.info.last_error = None; + } + } + RebalStatus::Completed => { + if !matches!(remote.info.status, RebalStatus::Failed | RebalStatus::Stopped) { + remote.info.status = RebalStatus::Completed; + remote.info.end_time = local.info.end_time.or(remote.info.end_time); + remote.info.last_error = None; + } + } + RebalStatus::Started => { + if !is_rebalance_terminal_status(remote.info.status) { + remote.info.status = RebalStatus::Started; + remote.info.last_error = local.info.last_error.clone(); + } + } + RebalStatus::None => {} + } +} + +fn merge_rebalance_meta(remote: &mut RebalanceMeta, local: &RebalanceMeta) { + if remote.id.is_empty() { + *remote = local.clone(); + return; + } + + if !local.id.is_empty() && remote.id != local.id { + *remote = local.clone(); + return; + } + + remote.percent_free_goal = local.percent_free_goal; + remote.last_refreshed_at = Some(OffsetDateTime::now_utc()); + if remote.stopped_at.is_none() { + remote.stopped_at = local.stopped_at; + } + + if remote.pool_stats.len() < local.pool_stats.len() { + remote.pool_stats.resize_with(local.pool_stats.len(), RebalanceStats::default); + } + + for (idx, local_pool_stat) in local.pool_stats.iter().enumerate() { + if let Some(remote_pool_stat) = remote.pool_stats.get_mut(idx) { + merge_rebalance_pool_stats(remote_pool_stat, local_pool_stat); + } + } +} + fn mark_started_rebalance_pools_stopped(meta: &mut RebalanceMeta, stop_time: OffsetDateTime) { for pool_stat in meta.pool_stats.iter_mut() { if pool_stat.info.status == RebalStatus::Started { @@ -1572,7 +2034,7 @@ impl ECStore { set: Arc, bucket_configs: Arc, // wk: Arc, - ) -> Result<()> { + ) -> Result { info!("rebalance_entry: start rebalance_entry"); // defer!(|| async { @@ -1583,12 +2045,12 @@ impl ECStore { if entry.is_dir() { info!("rebalance_entry: entry is dir, skipping"); - return Ok(()); + return Ok(RebalanceEntryOutcome::Completed); } if self.check_if_rebalance_done(pool_index).await { info!("rebalance_entry: rebalance done, skipping pool {}", pool_index); - return Ok(()); + return Ok(RebalanceEntryOutcome::Completed); } let mut fivs = @@ -1599,6 +2061,7 @@ impl ECStore { let mut rebalanced: usize = 0; let mut expired: usize = 0; + let mut stats_updates = Vec::with_capacity(fivs.versions.len()); for version in fivs.versions.iter() { if crate::pools::should_skip_lifecycle_for_data_movement( self.clone(), @@ -1664,20 +2127,53 @@ impl ECStore { "rebalance_entry {} Error rebalancing entry {}/{:?}: {:?}", &bucket, &version.name, &version.version_id, err ); - return Err(with_rebalance_entry_context("migrate", bucket.as_str(), version.name.as_str(), err)); + if should_defer_rebalance_entry_failure(&err) { + let deferred_error = format!("{REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX} {err}"); + warn!( + "rebalance_entry {} deferring transient migration failure for {}/{:?}: {}", + &bucket, &version.name, &version.version_id, err + ); + if let Err(stats_err) = self.update_rebalance_last_error(pool_index, deferred_error.clone()).await { + error!( + "rebalance_entry {} failed to record deferred transient failure for {}: {}", + &bucket, &entry.name, stats_err + ); + } + return Ok(RebalanceEntryOutcome::Deferred { + last_error: deferred_error, + }); + } + let entry_err = + with_rebalance_entry_context(result.stage.unwrap_or("migrate"), bucket.as_str(), version.name.as_str(), err); + + if !stats_updates.is_empty() + && let Err(stats_err) = self + .update_pool_stats_batch(pool_index, bucket.clone(), stats_updates.as_slice()) + .await + { + error!( + "rebalance_entry {} failed to update stats before returning migration error for {}: {}", + &bucket, &entry.name, stats_err + ); + } + + return Err(entry_err); } - resolve_rebalance_stats_update_result( - self.update_pool_stats(pool_index, bucket.clone(), version).await, - pool_index, - bucket.as_str(), - version.name.as_str(), - )?; + stats_updates.push(version); if should_count_rebalance_version_complete(&result) { rebalanced += 1; } } + resolve_rebalance_stats_update_result( + self.update_pool_stats_batch(pool_index, bucket.clone(), stats_updates.as_slice()) + .await, + pool_index, + bucket.as_str(), + entry.name.as_str(), + )?; + if should_cleanup_rebalance_source_entry(rebalanced, fivs.versions.len()) { resolve_rebalance_entry_cleanup_delete_result( set.delete_object( @@ -1693,11 +2189,12 @@ impl ECStore { .await, bucket.as_str(), entry.name.as_str(), - )?; + ) + .map_err(|err| with_rebalance_entry_context("cleanup_source", bucket.as_str(), entry.name.as_str(), err))?; info!("rebalance_entry {} Entry {} deleted successfully", &bucket, &entry.name); } - Ok(()) + Ok(RebalanceEntryOutcome::Completed) } #[tracing::instrument(skip(self, rd))] @@ -1705,8 +2202,29 @@ impl ECStore { data_movement::migrate_object(self, pool_idx, bucket, rd, "rebalance_object").await } + async fn update_rebalance_last_error(&self, pool_idx: usize, message: String) -> Result<()> { + let mut rebalance_meta = self.rebalance_meta.write().await; + let Some(meta) = rebalance_meta.as_mut() else { + return Err(rebalance_metadata_not_initialized_error("record rebalance last error")); + }; + let pool_count = meta.pool_stats.len(); + ensure_valid_rebalance_pool_index(pool_count, pool_idx)?; + let Some(pool_stat) = meta.pool_stats.get_mut(pool_idx) else { + return Err(invalid_rebalance_pool_index_error(pool_idx, pool_count)); + }; + + pool_stat.info.last_error = Some(message); + meta.last_refreshed_at = Some(OffsetDateTime::now_utc()); + Ok(()) + } + #[tracing::instrument(skip(self, rx))] - async fn rebalance_bucket(self: &Arc, rx: CancellationToken, bucket: String, pool_index: usize) -> Result<()> { + async fn rebalance_bucket( + self: &Arc, + rx: CancellationToken, + bucket: String, + pool_index: usize, + ) -> Result { ensure_valid_rebalance_pool_index(self.pools.len(), pool_index)?; // Placeholder for actual bucket rebalance logic @@ -1794,7 +2312,15 @@ impl ECStore { let entry_tasks = entry_tasks.clone(); let job = tokio::spawn(async move { - let list_result = set.list_objects_to_rebalance(rx, bucket, rebalance_entry).await; + let list_result = run_rebalance_listing_with_retry( + set, + rx, + bucket.clone(), + rebalance_entry, + set_idx, + REBALANCE_LISTING_MAX_ATTEMPTS, + ) + .await; let entry_result = wait_rebalance_entry_tasks(set_idx, entry_tasks).await; let result = list_result.and(entry_result); if let Err(err) = &result { @@ -1809,18 +2335,27 @@ impl ECStore { } let mut worker_error: Option = None; + let mut deferred_error: Option = None; for (set_idx, job) in jobs { - if let Err(err) = resolve_rebalance_worker_result(set_idx, job.await) - && worker_error.is_none() - { - worker_error = Some(err); + match resolve_rebalance_worker_result(set_idx, job.await) { + Ok(Some(last_error)) if deferred_error.is_none() => { + deferred_error = Some(last_error); + } + Ok(_) => {} + Err(err) if worker_error.is_none() => { + worker_error = Some(err); + } + Err(_) => {} } } let entry_error = entry_error.lock().await.clone(); resolve_rebalance_bucket_error(entry_error, worker_error)?; + if let Some(last_error) = deferred_error { + return Ok(RebalanceBucketOutcome::Deferred { last_error }); + } info!("rebalance_bucket: rebalance_bucket done"); - Ok(()) + Ok(RebalanceBucketOutcome::Completed) } #[tracing::instrument(skip(self))] @@ -1843,12 +2378,66 @@ impl ECStore { pool_idx, opt, meta_to_save ); let stage = format!("save_rebalance_stats for pool {pool_idx} opt {opt:?}"); - resolve_rebalance_meta_save_result(meta_to_save.save(pool).await, stage.as_str())?; + resolve_rebalance_meta_save_result( + self.save_rebalance_meta_with_merge(pool, &meta_to_save, stage.as_str()).await, + stage.as_str(), + )?; Ok(()) } } +async fn run_rebalance_listing_with_retry( + set: Arc, + rx: CancellationToken, + bucket: String, + cb: ListCallback, + set_idx: usize, + max_attempts: usize, +) -> Result<()> { + let max_attempts = max_attempts.max(1); + let mut last_error = None; + + for attempt in 0..max_attempts { + match set.list_objects_to_rebalance(rx.clone(), bucket.clone(), cb.clone()).await { + Ok(()) => return Ok(()), + Err(err) if should_retry_rebalance_listing(&err, attempt, max_attempts) => { + let next_attempt = attempt + 2; + let delay = rebalance_listing_retry_delay(attempt); + error!( + "rebalance listing failed for bucket {} set {} attempt {}/{}: {}; retrying in {:?}", + bucket, + set_idx, + attempt + 1, + max_attempts, + err, + delay + ); + last_error = Some(err); + wait_rebalance_listing_retry(&rx, delay).await?; + info!( + "rebalance listing retrying bucket {} set {} attempt {}/{}", + bucket, set_idx, next_attempt, max_attempts + ); + } + Err(err) => { + return Err(Error::other(format!( + "rebalance listing failed for bucket {bucket} set {set_idx} attempt {}/{}: {err}", + attempt + 1, + max_attempts + ))); + } + } + } + + Err(Error::other(format!( + "rebalance listing failed for bucket {bucket} set {set_idx} after {max_attempts} attempts: {}", + last_error + .map(|err| err.to_string()) + .unwrap_or_else(|| "unknown listing failure".to_string()) + ))) +} + impl SetDisks { #[tracing::instrument(skip(self, rx, cb))] pub async fn list_objects_to_rebalance( @@ -1880,6 +2469,7 @@ impl SetDisks { bucket: bucket.clone(), recursive: true, min_disks: listing_quorum, + skip_walkdir_total_timeout: true, agreed: Some(Box::new(move |entry: MetaCacheEntry| { info!("list_objects_to_rebalance: agreed: {:?}", &entry.name); Box::pin(cb1(entry)) @@ -1912,6 +2502,7 @@ impl SetDisks { #[cfg(test)] mod rebalance_unit_tests { + use super::REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX; use super::first_rebalance_bucket; use super::is_rebalance_actively_running; use super::is_rebalance_conflicting_with_decommission; @@ -1919,23 +2510,27 @@ mod rebalance_unit_tests { use super::percent_free_ratio; use super::rebalance_goal_reached; use super::{ - GetObjectReader, HTTPRangeSpec, MigrationBackend, MigrationVersionResult, ObjectInfo, ObjectOptions, RebalSaveOpt, - RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats, RebalanceTerminalEvent, apply_rebalance_save_option, - apply_rebalance_terminal_event, apply_stopped_at, classify_rebalance_terminal_event, clone_arc_by_index, clone_first_arc, - clone_rebalance_pool_stats, complete_rebalance_pools_at_goal, ensure_rebalance_listing_disks_available, - ensure_rebalance_not_decommissioning, ensure_valid_rebalance_pool_index, is_rebalance_stopped_terminal_event, - load_rebalance_bucket_configs, mark_rebalance_bucket_done, migrate_entry_version, next_rebal_bucket_from_stat, - rebalance_delete_marker_opts, rebalance_meta_load_no_data_error, rebalance_meta_load_unknown_format_error, - rebalance_meta_load_unknown_version_error, resolve_load_rebalance_stats_update_result, resolve_next_rebalance_bucket, + DiskError, GetObjectReader, HTTPRangeSpec, MigrationBackend, MigrationVersionResult, ObjectInfo, ObjectOptions, + RebalSaveOpt, RebalStatus, RebalanceBucketOutcome, RebalanceEntryOutcome, RebalanceInfo, RebalanceMeta, RebalanceStats, + RebalanceTerminalEvent, apply_rebalance_save_option, apply_rebalance_terminal_event, apply_stopped_at, + classify_rebalance_terminal_event, clone_arc_by_index, clone_first_arc, clone_rebalance_pool_stats, + complete_rebalance_pools_at_goal, defer_bucket_in_rebalance_queue, ensure_rebalance_listing_disks_available, + ensure_rebalance_not_decommissioning, ensure_valid_rebalance_pool_index, has_deferred_rebalance_error, + is_rebalance_stopped_terminal_event, is_transient_rebalance_error, load_rebalance_bucket_configs, + mark_rebalance_bucket_done, merge_rebalance_meta, migrate_entry_version, migrate_entry_version_with_retry_wait, + next_rebal_bucket_from_stat, rebalance_delete_marker_opts, rebalance_listing_retry_delay, + rebalance_meta_load_no_data_error, rebalance_meta_load_unknown_format_error, rebalance_meta_load_unknown_version_error, + rebalance_migration_retry_delay, resolve_load_rebalance_stats_update_result, resolve_next_rebalance_bucket, resolve_rebalance_bucket_error, resolve_rebalance_bucket_result, resolve_rebalance_entry_cleanup_delete_result, resolve_rebalance_file_info_versions_result, resolve_rebalance_meta_load_result, resolve_rebalance_meta_save_result, resolve_rebalance_migrate_result_error, resolve_rebalance_optional_bucket_config_result, resolve_rebalance_participants, resolve_rebalance_save_task_result, resolve_rebalance_stats_update_result, resolve_rebalance_terminal_error, - resolve_rebalance_worker_result, send_rebalance_done_signal, should_cleanup_rebalance_source_entry, - should_count_rebalance_version_complete, should_ignore_rebalance_data_usage_cache, should_pool_participate, - should_preserve_rebalance_stopped_state, should_skip_rebalance_delete_marker, should_skip_start_rebalance, + resolve_rebalance_worker_result, send_rebalance_done_signal, should_accept_rebalance_stats_update, + should_cleanup_rebalance_source_entry, should_count_rebalance_version_complete, should_defer_rebalance_entry_failure, + should_ignore_rebalance_data_usage_cache, should_pool_participate, should_preserve_rebalance_stopped_state, + should_retry_rebalance_listing, should_skip_rebalance_delete_marker, should_skip_start_rebalance, stop_rebalance_meta_snapshot, stop_rebalance_state, take_bucket_from_rebalance_queue, validate_start_rebalance_state, - with_rebalance_entry_context, + wait_rebalance_listing_retry, with_rebalance_entry_context, }; use crate::data_movement; use crate::data_usage::DATA_USAGE_CACHE_NAME; @@ -1951,6 +2546,7 @@ mod rebalance_unit_tests { use std::sync::atomic::{AtomicUsize, Ordering}; use time::OffsetDateTime; use tokio::sync::mpsc; + use tokio::time::Duration; use tokio_util::sync::CancellationToken; struct MigrationBackendSpy { @@ -2173,6 +2769,41 @@ mod rebalance_unit_tests { assert_eq!(backend.delete_calls(), 0); } + #[tokio::test] + async fn test_migrate_entry_version_remote_overwrite_is_not_ignored() { + let backend = MigrationBackendSpy::new( + None, + None, + Some(Err(Error::DataMovementOverwriteErr( + "bucket".to_string(), + "object.bin".to_string(), + "vid-1".to_string(), + ))), + ); + let version = version_remote(); + let mut transfer = |_, _, _| async move { Ok(()) }; + + let result = migrate_entry_version( + &backend, + "bucket".to_string(), + 0, + &version, + Some("vid-1".to_string()), + 3, + false, + &mut transfer, + ) + .await; + + assert!(result.failed); + assert!(!result.ignored); + assert!(!result.cleanup_ignored); + assert!(!result.moved); + assert_eq!(result.stage, Some("move_remote_version")); + assert!(matches!(result.error, Some(Error::DataMovementOverwriteErr(_, _, _)))); + assert_eq!(backend.move_remote_calls(), 1); + } + #[tokio::test] async fn test_migrate_entry_version_remote_failure_is_reported() { let backend = MigrationBackendSpy::new(None, Some(Ok(ObjectInfo::default())), Some(Err(Error::SlowDown))); @@ -2269,6 +2900,41 @@ mod rebalance_unit_tests { assert_eq!(backend.delete_calls(), 1); } + #[tokio::test] + async fn test_migrate_entry_version_deleted_version_overwrite_is_not_ignored() { + let backend = MigrationBackendSpy::new( + None, + Some(Err(Error::DataMovementOverwriteErr( + "bucket".to_string(), + "object.bin".to_string(), + "vid-1".to_string(), + ))), + None, + ); + let version = version_deleted(); + let mut transfer = |_, _, _| async move { Ok(()) }; + + let result = migrate_entry_version( + &backend, + "bucket".to_string(), + 1, + &version, + Some("vid-1".to_string()), + 3, + false, + &mut transfer, + ) + .await; + + assert!(result.failed); + assert!(!result.ignored); + assert!(!result.cleanup_ignored); + assert!(!result.moved); + assert_eq!(result.stage, Some("delete_marker")); + assert!(matches!(result.error, Some(Error::DataMovementOverwriteErr(_, _, _)))); + assert_eq!(backend.delete_calls(), 1); + } + #[tokio::test] async fn test_migrate_entry_version_reader_not_found_is_ignored() { let backend = MigrationBackendSpy::new( @@ -2304,6 +2970,7 @@ mod rebalance_unit_tests { async fn test_migrate_entry_version_reader_retries_before_success() { let backend = MigrationBackendSpy::new(Some(Err(Error::SlowDown)), None, None); let transfer_count = Arc::new(AtomicUsize::new(0)); + let wait_count = Arc::new(AtomicUsize::new(0)); let mut transfer = { let transfer_count = transfer_count.clone(); move |_, _, _| { @@ -2316,7 +2983,7 @@ mod rebalance_unit_tests { }; let version = version_normal(); - let result = migrate_entry_version( + let result = migrate_entry_version_with_retry_wait( &backend, "bucket".to_string(), 1, @@ -2325,6 +2992,15 @@ mod rebalance_unit_tests { 3, false, &mut transfer, + { + let wait_count = wait_count.clone(); + move |_| { + let wait_count = wait_count.clone(); + async move { + wait_count.fetch_add(1, Ordering::SeqCst); + } + } + }, ) .await; @@ -2336,6 +3012,7 @@ mod rebalance_unit_tests { assert_eq!(backend.get_calls(), 2); assert_eq!(backend.delete_calls(), 0); assert_eq!(transfer_count.load(Ordering::SeqCst), 1); + assert_eq!(wait_count.load(Ordering::SeqCst), 1); } struct AlwaysFailGetBackend { @@ -2461,6 +3138,7 @@ mod rebalance_unit_tests { async fn test_migrate_entry_version_transfer_retries_before_success() { let backend = MigrationBackendSpy::new(Some(Ok(MigrationBackendSpy::make_reader())), None, None); let transfer_count = Arc::new(AtomicUsize::new(0)); + let wait_count = Arc::new(AtomicUsize::new(0)); let mut transfer = { let transfer_count = transfer_count.clone(); move |_, _, _| { @@ -2476,7 +3154,7 @@ mod rebalance_unit_tests { }; let version = version_normal(); - let result = migrate_entry_version( + let result = migrate_entry_version_with_retry_wait( &backend, "bucket".to_string(), 1, @@ -2485,6 +3163,15 @@ mod rebalance_unit_tests { 3, false, &mut transfer, + { + let wait_count = wait_count.clone(); + move |_| { + let wait_count = wait_count.clone(); + async move { + wait_count.fetch_add(1, Ordering::SeqCst); + } + } + }, ) .await; @@ -2494,6 +3181,54 @@ mod rebalance_unit_tests { assert!(!result.failed); assert_eq!(backend.get_calls(), 2); assert_eq!(transfer_count.load(Ordering::SeqCst), 2); + assert_eq!(wait_count.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn test_migrate_entry_version_transfer_non_transient_fails_without_retry() { + let backend = MigrationBackendSpy::new(Some(Ok(MigrationBackendSpy::make_reader())), None, None); + let transfer_count = Arc::new(AtomicUsize::new(0)); + let wait_count = Arc::new(AtomicUsize::new(0)); + let mut transfer = { + let transfer_count = transfer_count.clone(); + move |_, _, _| { + let transfer_count = transfer_count.clone(); + async move { + transfer_count.fetch_add(1, Ordering::SeqCst); + Err(Error::FileAccessDenied) + } + } + }; + + let version = version_normal(); + let result = migrate_entry_version_with_retry_wait( + &backend, + "bucket".to_string(), + 1, + &version, + version.version_id.map(|v| v.to_string()), + 3, + false, + &mut transfer, + { + let wait_count = wait_count.clone(); + move |_| { + let wait_count = wait_count.clone(); + async move { + wait_count.fetch_add(1, Ordering::SeqCst); + } + } + }, + ) + .await; + + assert!(result.failed); + assert!(!result.ignored); + assert_eq!(result.stage, Some("write_target")); + assert!(matches!(result.error, Some(Error::FileAccessDenied))); + assert_eq!(backend.get_calls(), 1); + assert_eq!(transfer_count.load(Ordering::SeqCst), 1); + assert_eq!(wait_count.load(Ordering::SeqCst), 0); } #[tokio::test] @@ -2506,7 +3241,7 @@ mod rebalance_unit_tests { let transfer_count = transfer_count.clone(); async move { transfer_count.fetch_add(1, Ordering::SeqCst); - Err(Error::NotModified) + Err(Error::SlowDown) } } }; @@ -2569,6 +3304,47 @@ mod rebalance_unit_tests { assert_eq!(transfer_count.load(Ordering::SeqCst), 1); } + #[tokio::test] + async fn test_migrate_entry_version_transfer_overwrite_is_not_ignored() { + let backend = MigrationBackendSpy::new(Some(Ok(MigrationBackendSpy::make_reader())), None, None); + let transfer_count = Arc::new(AtomicUsize::new(0)); + let mut transfer = { + let transfer_count = transfer_count.clone(); + move |_, _, _| { + let transfer_count = transfer_count.clone(); + async move { + transfer_count.fetch_add(1, Ordering::SeqCst); + Err(Error::DataMovementOverwriteErr( + "bucket".to_string(), + "object.bin".to_string(), + "vid-1".to_string(), + )) + } + } + }; + + let version = version_normal(); + let result = migrate_entry_version( + &backend, + "bucket".to_string(), + 1, + &version, + Some("vid-1".to_string()), + 3, + false, + &mut transfer, + ) + .await; + + assert!(result.failed); + assert!(!result.ignored); + assert!(!result.cleanup_ignored); + assert!(!result.moved); + assert_eq!(result.stage, Some("write_target")); + assert!(matches!(result.error, Some(Error::DataMovementOverwriteErr(_, _, _)))); + assert_eq!(transfer_count.load(Ordering::SeqCst), 1); + } + #[tokio::test] async fn test_migrate_entry_version_ignores_data_usage_cache_when_enabled() { let backend = MigrationBackendSpy::new(Some(Ok(MigrationBackendSpy::make_reader())), None, None); @@ -2712,7 +3488,7 @@ mod rebalance_unit_tests { fn test_resolve_rebalance_worker_result_passthrough() { assert!(resolve_rebalance_worker_result(0, Ok(Ok(()))).is_ok()); - let err = resolve_rebalance_worker_result(0, Ok(Err(Error::OperationCanceled))).unwrap_err(); + let err = resolve_rebalance_worker_result::<()>(0, Ok(Err(Error::OperationCanceled))).unwrap_err(); assert!(matches!(err, Error::OperationCanceled)); } @@ -2724,23 +3500,27 @@ mod rebalance_unit_tests { .await .expect_err("panic task should return JoinError"); - let err = resolve_rebalance_worker_result(7, Err(join_error)).unwrap_err(); + let err = resolve_rebalance_worker_result::<()>(7, Err(join_error)).unwrap_err(); assert!(err.to_string().contains("rebalance worker 7 task join error")); } #[tokio::test] async fn test_wait_rebalance_entry_tasks_returns_ok_for_successful_tasks() { - let tasks = Arc::new(tokio::sync::Mutex::new(vec![tokio::spawn(async { Ok(()) })])); + let tasks = Arc::new(tokio::sync::Mutex::new(vec![tokio::spawn(async { + Ok(RebalanceEntryOutcome::Completed) + })])); - super::wait_rebalance_entry_tasks(1, tasks) + let result = super::wait_rebalance_entry_tasks(1, tasks) .await .expect("successful entry tasks should pass"); + + assert!(result.is_none()); } #[tokio::test] async fn test_wait_rebalance_entry_tasks_returns_first_task_error() { let tasks = Arc::new(tokio::sync::Mutex::new(vec![ - tokio::spawn(async { Ok(()) }), + tokio::spawn(async { Ok(RebalanceEntryOutcome::Completed) }), tokio::spawn(async { Err(Error::other("entry failed")) }), ])); @@ -2751,6 +3531,24 @@ mod rebalance_unit_tests { assert!(err.to_string().contains("entry failed")); } + #[tokio::test] + async fn test_wait_rebalance_entry_tasks_returns_deferred_error_without_failing() { + let tasks = Arc::new(tokio::sync::Mutex::new(vec![ + tokio::spawn(async { Ok(RebalanceEntryOutcome::Completed) }), + tokio::spawn(async { + Ok(RebalanceEntryOutcome::Deferred { + last_error: "deferred transient rebalance entry failure: timeout".to_string(), + }) + }), + ])); + + let result = super::wait_rebalance_entry_tasks(1, tasks) + .await + .expect("deferred transient entry should not fail worker"); + + assert_eq!(result.as_deref(), Some("deferred transient rebalance entry failure: timeout")); + } + #[test] fn test_resolve_rebalance_save_task_result_passthrough() { assert!(resolve_rebalance_save_task_result(0, Ok(Ok(()))).is_ok()); @@ -2777,6 +3575,239 @@ mod rebalance_unit_tests { assert!(message.contains(Error::SlowDown.to_string().as_str())); } + #[test] + fn test_merge_rebalance_meta_preserves_updates_from_multiple_pools() { + let start_time = OffsetDateTime::from_unix_timestamp(1_000).unwrap(); + let mut remote = RebalanceMeta { + id: "rebal-1".to_string(), + percent_free_goal: 0.5, + pool_stats: vec![ + RebalanceStats { + buckets: vec!["bucket-a".to_string()], + participating: true, + info: RebalanceInfo { + start_time: Some(start_time), + status: RebalStatus::Started, + ..Default::default() + }, + num_versions: 4, + bytes: 400, + bucket: "bucket-a".to_string(), + object: "remote-object".to_string(), + ..Default::default() + }, + RebalanceStats { + buckets: vec!["bucket-a".to_string()], + participating: true, + info: RebalanceInfo { + start_time: Some(start_time), + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }, + ], + ..Default::default() + }; + let local = RebalanceMeta { + id: "rebal-1".to_string(), + percent_free_goal: 0.5, + pool_stats: vec![ + RebalanceStats { + buckets: vec!["bucket-a".to_string()], + participating: true, + info: RebalanceInfo { + start_time: Some(start_time), + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }, + RebalanceStats { + buckets: Vec::new(), + rebalanced_buckets: vec!["bucket-a".to_string()], + participating: true, + info: RebalanceInfo { + start_time: Some(start_time), + status: RebalStatus::Started, + ..Default::default() + }, + num_versions: 7, + bytes: 700, + bucket: "bucket-a".to_string(), + object: "local-object".to_string(), + ..Default::default() + }, + ], + ..Default::default() + }; + + merge_rebalance_meta(&mut remote, &local); + + assert_eq!(remote.pool_stats[0].num_versions, 4); + assert_eq!(remote.pool_stats[0].object, "remote-object"); + assert_eq!(remote.pool_stats[1].num_versions, 7); + assert_eq!(remote.pool_stats[1].object, "local-object"); + assert!(remote.pool_stats[1].buckets.is_empty()); + assert_eq!(remote.pool_stats[1].rebalanced_buckets, vec!["bucket-a"]); + } + + #[test] + fn test_merge_rebalance_meta_does_not_overwrite_failed_with_started_stats() { + let now = OffsetDateTime::from_unix_timestamp(2_000).unwrap(); + let mut remote = RebalanceMeta { + id: "rebal-1".to_string(), + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + status: RebalStatus::Failed, + end_time: Some(now), + last_error: Some("timeout".to_string()), + ..Default::default() + }, + num_versions: 10, + bytes: 1_000, + ..Default::default() + }], + ..Default::default() + }; + let local = RebalanceMeta { + id: "rebal-1".to_string(), + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + num_versions: 8, + bytes: 800, + ..Default::default() + }], + ..Default::default() + }; + + merge_rebalance_meta(&mut remote, &local); + + assert_eq!(remote.pool_stats[0].info.status, RebalStatus::Failed); + assert_eq!(remote.pool_stats[0].info.last_error.as_deref(), Some("timeout")); + assert_eq!(remote.pool_stats[0].num_versions, 10); + assert_eq!(remote.pool_stats[0].bytes, 1_000); + } + + #[test] + fn test_merge_rebalance_meta_does_not_overwrite_stopped_with_started_stats() { + let stopped_at = OffsetDateTime::from_unix_timestamp(3_000).unwrap(); + let mut remote = RebalanceMeta { + id: "rebal-1".to_string(), + stopped_at: Some(stopped_at), + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + status: RebalStatus::Stopped, + end_time: Some(stopped_at), + ..Default::default() + }, + num_versions: 5, + ..Default::default() + }], + ..Default::default() + }; + let local = RebalanceMeta { + id: "rebal-1".to_string(), + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + num_versions: 9, + ..Default::default() + }], + ..Default::default() + }; + + merge_rebalance_meta(&mut remote, &local); + + assert_eq!(remote.stopped_at, Some(stopped_at)); + assert_eq!(remote.pool_stats[0].info.status, RebalStatus::Stopped); + assert_eq!(remote.pool_stats[0].info.end_time, Some(stopped_at)); + assert_eq!(remote.pool_stats[0].num_versions, 9); + } + + #[test] + fn test_merge_rebalance_meta_preserves_failed_status_over_stopped() { + let stopped_at = OffsetDateTime::from_unix_timestamp(3_000).unwrap(); + let failed_at = OffsetDateTime::from_unix_timestamp(4_000).unwrap(); + let mut remote = RebalanceMeta { + id: "rebal-1".to_string(), + stopped_at: Some(stopped_at), + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + status: RebalStatus::Stopped, + end_time: Some(stopped_at), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + let local = RebalanceMeta { + id: "rebal-1".to_string(), + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + status: RebalStatus::Failed, + end_time: Some(failed_at), + last_error: Some("late failure".to_string()), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + + merge_rebalance_meta(&mut remote, &local); + + assert_eq!(remote.pool_stats[0].info.status, RebalStatus::Failed); + assert_eq!(remote.pool_stats[0].info.end_time, Some(failed_at)); + assert_eq!(remote.pool_stats[0].info.last_error.as_deref(), Some("late failure")); + } + + #[test] + fn test_merge_rebalance_meta_does_not_overwrite_stopped_with_completed_status() { + let stopped_at = OffsetDateTime::from_unix_timestamp(3_000).unwrap(); + let completed_at = OffsetDateTime::from_unix_timestamp(5_000).unwrap(); + let mut remote = RebalanceMeta { + id: "rebal-1".to_string(), + stopped_at: Some(stopped_at), + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + status: RebalStatus::Stopped, + end_time: Some(stopped_at), + ..Default::default() + }, + num_versions: 5, + ..Default::default() + }], + ..Default::default() + }; + let local = RebalanceMeta { + id: "rebal-1".to_string(), + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + status: RebalStatus::Completed, + end_time: Some(completed_at), + ..Default::default() + }, + num_versions: 12, + ..Default::default() + }], + ..Default::default() + }; + + merge_rebalance_meta(&mut remote, &local); + + assert_eq!(remote.stopped_at, Some(stopped_at)); + assert_eq!(remote.pool_stats[0].info.status, RebalStatus::Stopped); + assert_eq!(remote.pool_stats[0].info.end_time, Some(stopped_at)); + assert_eq!(remote.pool_stats[0].num_versions, 12); + } + #[test] fn test_rebalance_meta_load_no_data_error_formats_context() { let err = rebalance_meta_load_no_data_error(); @@ -2910,7 +3941,9 @@ mod rebalance_unit_tests { #[test] fn test_resolve_rebalance_bucket_result_passthrough() { - assert!(resolve_rebalance_bucket_result(Ok(()), 2, "bucket-a").is_ok()); + let outcome = resolve_rebalance_bucket_result(Ok(RebalanceBucketOutcome::Completed), 2, "bucket-a") + .expect("completed bucket should pass through"); + assert_eq!(outcome, RebalanceBucketOutcome::Completed); } #[test] @@ -2944,6 +3977,111 @@ mod rebalance_unit_tests { assert!(ensure_rebalance_listing_disks_available(true, "bucket-a").is_ok()); } + #[test] + fn test_is_transient_rebalance_error_classifies_retryable_errors() { + assert!(is_transient_rebalance_error(&Error::SlowDown)); + assert!(is_transient_rebalance_error(&Error::ErasureReadQuorum)); + assert!(is_transient_rebalance_error(&Error::ErasureWriteQuorum)); + assert!(is_transient_rebalance_error(&Error::Io(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "timed out", + )))); + assert!(is_transient_rebalance_error(&Error::other("i/o timeout"))); + } + + #[test] + fn test_is_transient_rebalance_error_accepts_lock_and_rpc_timeouts() { + assert!(is_transient_rebalance_error(&Error::Io(std::io::Error::other( + "Failed to acquire read lock: ns_loc: read lock acquisition timed out on bucket/object" + )))); + assert!(is_transient_rebalance_error(&Error::other( + "Remote lock RPC timed out: RPC timed out after 50ms" + ))); + assert!(is_transient_rebalance_error(&Error::other( + "Unknown hyper error: hyper::Error(Http2, KeepAliveTimedOut)" + ))); + assert!(is_transient_rebalance_error(&Error::Lock(rustfs_lock::LockError::timeout( + "bucket/object", + Duration::from_secs(5), + )))); + } + + #[test] + fn test_is_transient_rebalance_error_accepts_wrapped_disk_timeout() { + assert!(is_transient_rebalance_error(&Error::Io(std::io::Error::other(DiskError::Timeout)))); + } + + #[test] + fn test_is_transient_rebalance_error_accepts_io_timeout_message() { + assert!(is_transient_rebalance_error(&Error::Io(std::io::Error::other("timeout")))); + } + + #[test] + fn test_is_transient_rebalance_error_rejects_terminal_errors() { + assert!(!is_transient_rebalance_error(&Error::DataMovementOverwriteErr( + "bucket".to_string(), + "object".to_string(), + "version".to_string(), + ))); + assert!(!is_transient_rebalance_error(&Error::ObjectNotFound( + "bucket".to_string(), + "object".to_string(), + ))); + assert!(!is_transient_rebalance_error(&Error::FileAccessDenied)); + assert!(!is_transient_rebalance_error(&Error::Lock(rustfs_lock::LockError::already_locked( + "bucket/object", + "owner-a", + )))); + assert!(!is_transient_rebalance_error(&Error::other( + "Failed to acquire read lock: ns_loc: read lock acquisition failed on bucket/object: permission denied" + ))); + } + + #[test] + fn test_should_retry_rebalance_listing_respects_attempt_limit_and_error_type() { + assert!(should_retry_rebalance_listing(&Error::SlowDown, 0, 3)); + assert!(should_retry_rebalance_listing(&Error::SlowDown, 1, 3)); + assert!(!should_retry_rebalance_listing(&Error::SlowDown, 2, 3)); + assert!(!should_retry_rebalance_listing(&Error::FileAccessDenied, 0, 3)); + } + + #[test] + fn test_rebalance_listing_retry_delay_scales_by_attempt() { + assert_eq!(rebalance_listing_retry_delay(0), Duration::from_millis(250)); + assert_eq!(rebalance_listing_retry_delay(1), Duration::from_millis(500)); + } + + #[test] + fn test_rebalance_migration_retry_delay_scales_for_non_lock_timeout() { + assert_eq!(rebalance_migration_retry_delay(0, &Error::SlowDown), Duration::from_millis(250)); + assert_eq!(rebalance_migration_retry_delay(1, &Error::SlowDown), Duration::from_millis(500)); + } + + #[test] + fn test_should_defer_rebalance_entry_failure_only_for_transient_errors() { + assert!(should_defer_rebalance_entry_failure(&Error::Io(std::io::Error::other( + "Failed to acquire write lock: ns_loc: write lock acquisition timed out on bucket/object" + )))); + assert!(!should_defer_rebalance_entry_failure(&Error::DataMovementOverwriteErr( + "bucket".to_string(), + "object".to_string(), + "version".to_string(), + ))); + assert!(!should_defer_rebalance_entry_failure(&Error::FileAccessDenied)); + } + + #[tokio::test] + async fn test_wait_rebalance_listing_retry_returns_canceled_without_sleeping() { + let token = CancellationToken::new(); + token.cancel(); + + let err = wait_rebalance_listing_retry(&token, Duration::from_secs(30)) + .await + .expect_err("canceled rebalance should not wait before retrying"); + + assert!(matches!(err, Error::OperationCanceled)); + } + #[test] fn test_with_rebalance_entry_context_formats_stage_bucket_and_object() { let err = with_rebalance_entry_context("migrate", "bucket-a", "obj.txt", Error::SlowDown); @@ -2952,6 +4090,60 @@ mod rebalance_unit_tests { assert!(message.contains("Please reduce your request rate")); } + #[test] + fn test_with_rebalance_entry_context_formats_precise_stage() { + let err = with_rebalance_entry_context("write_target", "bucket-a", "obj.txt", Error::SlowDown); + let message = err.to_string(); + assert!(message.contains("rebalance entry write_target failed for bucket-a/obj.txt")); + assert!(message.contains("Please reduce your request rate")); + } + + #[tokio::test] + async fn test_migrate_entry_version_transfer_failure_reports_write_target_stage() { + let backend = MigrationBackendSpy::new(Some(Ok(MigrationBackendSpy::make_reader())), None, None); + let mut transfer = |_, _, _| async { Err(Error::SlowDown) }; + let version = version_normal(); + + let result = migrate_entry_version( + &backend, + "bucket".to_string(), + 1, + &version, + version.version_id.map(|v| v.to_string()), + 1, + false, + &mut transfer, + ) + .await; + + assert!(result.failed); + assert_eq!(result.stage, Some("write_target")); + assert!(matches!(result.error, Some(Error::SlowDown))); + } + + #[tokio::test] + async fn test_migrate_entry_version_reader_failure_reports_read_source_stage() { + let backend = AlwaysFailGetBackend::new(); + let mut transfer = |_, _, _| async { Ok(()) }; + let version = version_normal(); + + let result = migrate_entry_version( + &backend, + "bucket".to_string(), + 1, + &version, + version.version_id.map(|v| v.to_string()), + 1, + false, + &mut transfer, + ) + .await; + + assert!(result.failed); + assert_eq!(result.stage, Some("read_source")); + assert!(matches!(result.error, Some(Error::SlowDown))); + } + #[test] fn test_should_count_rebalance_version_complete_for_cleanup_safe_ignored_result() { let result = MigrationVersionResult { @@ -2996,6 +4188,58 @@ mod rebalance_unit_tests { assert!(!should_count_rebalance_version_complete(&MigrationVersionResult::default())); } + #[test] + fn test_should_accept_rebalance_stats_update_only_for_started_pool() { + let meta = RebalanceMeta { + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + + assert!(should_accept_rebalance_stats_update(&meta, 0)); + } + + #[test] + fn test_should_accept_rebalance_stats_update_rejects_stopped_meta() { + let meta = RebalanceMeta { + stopped_at: Some(OffsetDateTime::UNIX_EPOCH), + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + + assert!(!should_accept_rebalance_stats_update(&meta, 0)); + } + + #[test] + fn test_should_accept_rebalance_stats_update_rejects_terminal_or_invalid_pool() { + for status in [RebalStatus::Completed, RebalStatus::Stopped, RebalStatus::Failed] { + let meta = RebalanceMeta { + pool_stats: vec![RebalanceStats { + info: RebalanceInfo { + status, + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + + assert!(!should_accept_rebalance_stats_update(&meta, 0)); + assert!(!should_accept_rebalance_stats_update(&meta, 1)); + } + } + #[test] fn test_should_skip_rebalance_delete_marker_when_last_remaining_without_replication() { assert!(should_skip_rebalance_delete_marker(&version_deleted(), 1, false)); @@ -3402,6 +4646,31 @@ mod rebalance_unit_tests { assert_eq!(meta.pool_stats[1].info.status, RebalStatus::Started); } + #[test] + fn test_complete_rebalance_pools_at_goal_skips_deferred_transient_error() { + let now = OffsetDateTime::from_unix_timestamp(1_000).unwrap(); + let mut meta = RebalanceMeta { + percent_free_goal: 0.5, + pool_stats: vec![RebalanceStats { + participating: true, + init_free_space: 400, + init_capacity: 1_000, + bytes: 100, + info: RebalanceInfo { + status: RebalStatus::Started, + last_error: Some(format!("{REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX} timeout")), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + + assert!(!complete_rebalance_pools_at_goal(&mut meta, now)); + assert_eq!(meta.pool_stats[0].info.status, RebalStatus::Started); + assert!(has_deferred_rebalance_error(&meta.pool_stats[0])); + } + #[test] fn test_should_skip_start_rebalance_only_when_running_and_cancel_attached() { assert!(should_skip_start_rebalance(true, true)); @@ -3449,6 +4718,47 @@ mod rebalance_unit_tests { assert_eq!(decoded.total_compressed, 2_097_152); } + #[test] + fn test_merge_rebalance_bucket_lists_keeps_existing_order_and_skips_duplicates() { + let mut remote = vec!["bucket-a".to_string(), "bucket-b".to_string()]; + let local = vec![ + "bucket-b".to_string(), + "bucket-c".to_string(), + "bucket-c".to_string(), + "bucket-d".to_string(), + ]; + + super::merge_rebalance_bucket_lists(&mut remote, &local); + + assert_eq!( + remote, + vec![ + "bucket-a".to_string(), + "bucket-b".to_string(), + "bucket-c".to_string(), + "bucket-d".to_string() + ] + ); + } + + #[test] + fn test_remove_rebalanced_buckets_from_queue_filters_membership_linearly() { + let mut pool_stat = RebalanceStats { + buckets: vec![ + "bucket-a".to_string(), + "bucket-b".to_string(), + "bucket-c".to_string(), + "bucket-b".to_string(), + ], + rebalanced_buckets: vec!["bucket-b".to_string(), "bucket-d".to_string()], + ..Default::default() + }; + + super::remove_rebalanced_buckets_from_queue(&mut pool_stat); + + assert_eq!(pool_stat.buckets, vec!["bucket-a".to_string(), "bucket-c".to_string()]); + } + #[test] fn test_resolve_rebalance_participants_respects_runtime_pool_count() { let now = OffsetDateTime::now_utc(); @@ -3776,6 +5086,38 @@ mod rebalance_unit_tests { assert!(pool_stat.rebalanced_buckets.is_empty()); } + #[test] + fn test_defer_bucket_in_rebalance_queue_moves_bucket_to_back() { + let mut pool_stat = RebalanceStats { + buckets: vec!["bucket-a".to_string(), "bucket-b".to_string(), "bucket-c".to_string()], + rebalanced_buckets: Vec::new(), + ..Default::default() + }; + + defer_bucket_in_rebalance_queue(&mut pool_stat, "bucket-a").expect("queued bucket should be deferred"); + + assert_eq!( + pool_stat.buckets, + vec!["bucket-b".to_string(), "bucket-c".to_string(), "bucket-a".to_string()] + ); + assert!(pool_stat.rebalanced_buckets.is_empty()); + } + + #[test] + fn test_defer_bucket_in_rebalance_queue_rejects_missing_bucket() { + let mut pool_stat = RebalanceStats { + buckets: vec!["bucket-a".to_string(), "bucket-b".to_string()], + rebalanced_buckets: Vec::new(), + ..Default::default() + }; + + let err = defer_bucket_in_rebalance_queue(&mut pool_stat, "bucket-c").expect_err("missing bucket should not be deferred"); + + assert!(err.to_string().contains("failed to defer rebalance bucket bucket-c")); + assert_eq!(pool_stat.buckets, vec!["bucket-a".to_string(), "bucket-b".to_string()]); + assert!(pool_stat.rebalanced_buckets.is_empty()); + } + #[test] fn test_mark_rebalance_bucket_done_rejects_missing_meta() { let err = mark_rebalance_bucket_done(None, 0, "bucket-a").expect_err("missing meta should fail"); @@ -3829,6 +5171,26 @@ mod rebalance_unit_tests { assert_eq!(meta.pool_stats[0].rebalanced_buckets, vec!["bucket-a".to_string()]); } + #[test] + fn test_mark_rebalance_bucket_done_clears_deferred_transient_error() { + let mut meta = RebalanceMeta { + pool_stats: vec![RebalanceStats { + buckets: vec!["bucket-a".to_string()], + info: RebalanceInfo { + last_error: Some(format!("{REBALANCE_DEFERRED_ENTRY_ERROR_PREFIX} timeout")), + ..Default::default() + }, + ..Default::default() + }], + ..Default::default() + }; + + mark_rebalance_bucket_done(Some(&mut meta), 0, "bucket-a") + .expect("bucket in queue should be marked done after deferred retry succeeds"); + + assert!(meta.pool_stats[0].info.last_error.is_none()); + } + #[test] fn test_apply_stopped_at_transitions_started_pools_only() { let now = OffsetDateTime::now_utc(); @@ -4109,6 +5471,30 @@ mod rebalance_unit_tests { assert_eq!(stat.bytes, expected_bytes); } + #[test] + fn test_rebalance_stats_update_batch_matches_repeated_updates() { + let mut latest = version_normal(); + latest.is_latest = true; + latest.size = 128; + + let mut historical = version_normal(); + historical.is_latest = false; + historical.size = 64; + + let mut repeated = RebalanceStats::default(); + repeated.update("bucket-a".to_string(), &latest); + repeated.update("bucket-a".to_string(), &historical); + + let mut batched = RebalanceStats::default(); + batched.update_batch("bucket-a".to_string(), &[&latest, &historical]); + + assert_eq!(batched.bucket, repeated.bucket); + assert_eq!(batched.object, repeated.object); + assert_eq!(batched.num_objects, repeated.num_objects); + assert_eq!(batched.num_versions, repeated.num_versions); + assert_eq!(batched.bytes, repeated.bytes); + } + #[test] fn test_rebalance_stats_update_ignores_invalid_data_blocks() { let mut stat = RebalanceStats { diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index 829e323ac..f9b84787d 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -1282,6 +1282,11 @@ impl DiskAPI for RemoteDisk { let bucket = opts.bucket.clone(); let base_dir = opts.base_dir.clone(); let disk_for_log = disk.clone(); + let timeout_duration = if opts.skip_total_timeout { + Duration::ZERO + } else { + get_drive_walkdir_timeout() + }; self.execute_with_timeout_for_op_and_health_action( "walk_dir", @@ -1329,7 +1334,7 @@ impl DiskAPI for RemoteDisk { Err(last_err.unwrap_or_else(|| DiskError::other("walk_dir retry exhausted without captured error"))) }, - get_drive_walkdir_timeout(), + timeout_duration, FailureHealthAction::IgnoreFailure, ) .await @@ -2444,6 +2449,7 @@ mod tests { forward_to: None, limit: 10, disk_id: String::new(), + ..Default::default() }; let expected_body = serde_json::to_vec(&opts).unwrap(); let mut writer = Vec::new(); @@ -2463,6 +2469,37 @@ mod tests { } } + #[tokio::test] + async fn test_remote_disk_walk_dir_preserves_skip_total_timeout_option() { + let transport = RecordingInternodeDataTransport::default(); + let remote_disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await; + let opts = WalkDirOptions { + bucket: "bucket".to_string(), + base_dir: "prefix".to_string(), + recursive: true, + skip_total_timeout: true, + ..Default::default() + }; + let mut writer = Vec::new(); + + remote_disk + .walk_dir(opts, &mut writer) + .await + .expect("walk_dir should be sent through configured data transport"); + + let calls = transport.calls(); + assert_eq!(calls.len(), 1); + match &calls[0] { + RecordedTransportCall::WalkDir(request) => { + let sent_opts: WalkDirOptions = + serde_json::from_slice(&request.body).expect("walk_dir request body should deserialize"); + assert!(sent_opts.skip_total_timeout); + assert_eq!(request.stall_timeout, Some(get_drive_walkdir_stall_timeout())); + } + other => panic!("expected walk-dir transport call, got {other:?}"), + } + } + #[tokio::test] async fn test_remote_disk_walk_dir_retries_once_on_retryable_transport_error() { let transport = RetryingWalkDirInternodeDataTransport::with_steps(vec![ @@ -2479,6 +2516,7 @@ mod tests { forward_to: None, limit: 10, disk_id: String::new(), + ..Default::default() }; let mut writer = Vec::new(); @@ -2510,6 +2548,7 @@ mod tests { forward_to: None, limit: 10, disk_id: String::new(), + ..Default::default() }; let mut writer = Vec::new(); diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index dfeb1c143..ba36225fa 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -115,10 +115,24 @@ impl ECStore { object: &str, opts: &ObjectOptions, ) -> Result { + self.handle_new_multipart_upload_with_pool_idx(bucket, object, opts) + .await + .map(|(res, _)| res) + } + + pub(crate) async fn handle_new_multipart_upload_with_pool_idx( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + ) -> Result<(MultipartUploadResult, usize)> { check_new_multipart_args(bucket, object)?; if self.single_pool() { - return self.pools[0].new_multipart_upload(bucket, object, opts).await; + return self.pools[0] + .new_multipart_upload(bucket, object, opts) + .await + .map(|res| (res, 0)); } for (idx, pool) in self.pools.iter().enumerate() { @@ -130,7 +144,8 @@ impl ECStore { .await?; if !res.uploads.is_empty() { - return self.pools[idx].new_multipart_upload(bucket, object, opts).await; + let res = self.pools[idx].new_multipart_upload(bucket, object, opts).await?; + return Ok((res, idx)); } } let idx = self.get_pool_idx(bucket, object, -1).await?; @@ -142,7 +157,8 @@ impl ECStore { )); } - self.pools[idx].new_multipart_upload(bucket, object, opts).await + let res = self.pools[idx].new_multipart_upload(bucket, object, opts).await?; + Ok((res, idx)) } #[instrument(skip(self))] diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index b04a4acb0..2ddb33436 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -233,6 +233,89 @@ fn data_movement_pool_lookup_opts(opts: &ObjectOptions, no_lock: bool) -> Object lookup_opts } +fn effective_object_actual_size(info: &ObjectInfo) -> Option { + info.get_actual_size().ok() +} + +fn is_equivalent_data_movement_delete_marker(source: &ObjectInfo, target: &ObjectInfo) -> bool { + is_data_movement_delete_marker(source) + && is_data_movement_delete_marker(target) + && source.version_id == target.version_id + && source.mod_time == target.mod_time +} + +fn is_data_movement_delete_marker(info: &ObjectInfo) -> bool { + info.delete_marker +} + +fn is_equivalent_data_movement_tiered_object(source: &rustfs_filemeta::FileInfo, target: &ObjectInfo) -> bool { + source.version_id == target.version_id + && !target.delete_marker + && source.size == target.size + && source.get_etag() == target.etag + && source.checksum == target.checksum + && source.mod_time == target.mod_time + && source.transition_status == target.transitioned_object.status + && source.transitioned_objname == target.transitioned_object.name + && source.transition_tier == target.transitioned_object.tier + && source + .transition_version_id + .map(|version_id| version_id.to_string()) + .unwrap_or_default() + == target.transitioned_object.version_id + && effective_object_actual_size(target) == Some(source.size) +} + +fn should_check_data_movement_resume_target(src_pool_idx: usize, target_pool_idx: usize) -> bool { + target_pool_idx != src_pool_idx +} + +fn resolve_data_movement_resume_target_pool( + selected_target_pool_idx: usize, + resume_target_pool_idx: Option, + src_pool_idx: usize, +) -> usize { + if should_check_data_movement_resume_target(src_pool_idx, selected_target_pool_idx) { + selected_target_pool_idx + } else { + resume_target_pool_idx.unwrap_or(selected_target_pool_idx) + } +} + +fn resolve_data_movement_delete_marker_resume_result( + target_result: Result>, + source: &ObjectInfo, + src_pool_idx: usize, + target_pool_idx: usize, +) -> Result { + if !should_check_data_movement_resume_target(src_pool_idx, target_pool_idx) { + return Ok(false); + } + + let Some(target) = target_result? else { + return Ok(false); + }; + + Ok(is_equivalent_data_movement_delete_marker(source, &target)) +} + +fn resolve_data_movement_tiered_resume_result( + target_result: Result>, + source: &rustfs_filemeta::FileInfo, + src_pool_idx: usize, + target_pool_idx: usize, +) -> Result { + if !should_check_data_movement_resume_target(src_pool_idx, target_pool_idx) { + return Ok(false); + } + + let Some(target) = target_result? else { + return Ok(false); + }; + + Ok(is_equivalent_data_movement_tiered_object(source, &target)) +} + impl ECStore { fn map_namespace_lock_error(bucket: &str, object: &str, mode: &'static str, err: rustfs_lock::LockError) -> StorageError { match err { @@ -377,6 +460,62 @@ impl ECStore { } } + async fn find_data_movement_target_info( + &self, + bucket: &str, + object: &str, + target_pool_idx: usize, + opts: &ObjectOptions, + ) -> Result> { + let lookup_opts = version_aware_lookup_opts(opts, true); + + let Some(pool) = self.pools.get(target_pool_idx) else { + return Err(Error::other(format!( + "data movement resume target pool {target_pool_idx} is out of range for {bucket}/{object}" + ))); + }; + + match pool.get_object_info(bucket, object, &lookup_opts).await { + Ok(info) => Ok(Some(info)), + Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => Ok(None), + Err(err) => Err(err), + } + } + + async fn has_equivalent_data_movement_delete_marker( + &self, + bucket: &str, + object: &str, + source: &ObjectInfo, + opts: &ObjectOptions, + target_pool_idx: usize, + ) -> Result { + resolve_data_movement_delete_marker_resume_result( + self.find_data_movement_target_info(bucket, object, target_pool_idx, opts) + .await, + source, + opts.src_pool_idx, + target_pool_idx, + ) + } + + async fn has_equivalent_data_movement_tiered_object( + &self, + bucket: &str, + object: &str, + source: &rustfs_filemeta::FileInfo, + opts: &ObjectOptions, + target_pool_idx: usize, + ) -> Result { + resolve_data_movement_tiered_resume_result( + self.find_data_movement_target_info(bucket, object, target_pool_idx, opts) + .await, + source, + opts.src_pool_idx, + target_pool_idx, + ) + } + fn resolve_decommission_target_pool_idx_result(result: Result, bucket: &str, object: &str) -> Result { result.map_err(|err| Error::other(format!("failed to select decommission target pool for {bucket}/{object}: {err}"))) } @@ -419,6 +558,17 @@ impl ECStore { )? }; if opts.data_movement && idx == opts.src_pool_idx { + let resume_target_pool_idx = self + .get_available_pool_idx_excluding(bucket, &object, fi.size, opts.src_pool_idx) + .await; + let target_pool_idx = resolve_data_movement_resume_target_pool(idx, resume_target_pool_idx, opts.src_pool_idx); + if self + .has_equivalent_data_movement_tiered_object(bucket, &object, fi, opts, target_pool_idx) + .await? + { + return Ok(()); + } + return Err(StorageError::DataMovementOverwriteErr( bucket.to_owned(), object.to_owned(), @@ -666,17 +816,44 @@ impl ECStore { let gopts = version_aware_lookup_opts(&opts, true); if opts.data_movement { - let existing_pool_idx = self - .get_pool_info_existing_with_opts(bucket, object, &gopts) - .await - .map(|(pinfo, _)| pinfo.index); - let target_pool_idx = + let existing_pool_info = self.get_pool_info_existing_with_opts(bucket, object, &gopts).await; + let existing_pool_idx = existing_pool_info + .as_ref() + .map(|(pinfo, _)| pinfo.index) + .map_err(Clone::clone); + let selected_target_pool_idx = match select_data_movement_target_pool(existing_pool_idx, opts.src_pool_idx, opts.delete_marker)? { Some(pool_idx) => pool_idx, None => self.get_pool_idx_no_lock(bucket, object, 0).await?, }; + let resume_target_pool_idx = if selected_target_pool_idx == opts.src_pool_idx { + self.get_available_pool_idx_excluding(bucket, object, 0, opts.src_pool_idx) + .await + } else { + None + }; + let target_pool_idx = + resolve_data_movement_resume_target_pool(selected_target_pool_idx, resume_target_pool_idx, opts.src_pool_idx); + + if opts.src_pool_idx == selected_target_pool_idx { + if let Ok((source_pool_info, _)) = existing_pool_info + && opts.delete_marker + && is_data_movement_delete_marker(&source_pool_info.object_info) + && self + .has_equivalent_data_movement_delete_marker( + bucket, + object, + &source_pool_info.object_info, + &opts, + target_pool_idx, + ) + .await? + { + let mut obj = source_pool_info.object_info; + obj.name = decode_dir_object(object); + return Ok(obj); + } - if opts.src_pool_idx == target_pool_idx { return Err(StorageError::DataMovementOverwriteErr( bucket.to_owned(), object.to_owned(), @@ -684,7 +861,9 @@ impl ECStore { )); } - let mut obj = self.pools[target_pool_idx].delete_object(bucket, object, opts).await?; + let mut obj = self.pools[selected_target_pool_idx] + .delete_object(bucket, object, opts) + .await?; obj.name = decode_dir_object(obj.name.as_str()); return Ok(obj); } @@ -1086,6 +1265,9 @@ impl ECStore { #[cfg(test)] mod tests { use super::*; + use crate::bucket::lifecycle::bucket_lifecycle_ops::TransitionedObject; + use crate::bucket::lifecycle::core::TRANSITION_COMPLETE; + use bytes::Bytes; use std::io::Cursor; use tokio::io::AsyncReadExt; @@ -1108,6 +1290,240 @@ mod tests { assert_eq!(target, Some(0)); } + #[test] + fn equivalent_data_movement_delete_marker_requires_same_version_and_mod_time() { + let version_id = Uuid::nil(); + let mod_time = OffsetDateTime::UNIX_EPOCH; + let source = ObjectInfo { + version_id: Some(version_id), + delete_marker: true, + mod_time: Some(mod_time), + ..Default::default() + }; + let target = source.clone(); + + assert!(is_equivalent_data_movement_delete_marker(&source, &target)); + + let mismatched = ObjectInfo { + mod_time: Some(mod_time + Duration::from_secs(1)), + ..target + }; + assert!(!is_equivalent_data_movement_delete_marker(&source, &mismatched)); + } + + #[test] + fn equivalent_data_movement_delete_marker_rejects_live_object() { + let source = ObjectInfo { + delete_marker: true, + ..Default::default() + }; + let target = ObjectInfo { + delete_marker: false, + ..source.clone() + }; + + assert!(!is_equivalent_data_movement_delete_marker(&source, &target)); + } + + #[test] + fn data_movement_delete_marker_resume_accepts_equivalent_target() { + let source = ObjectInfo { + version_id: Some(Uuid::nil()), + delete_marker: true, + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + + let should_resume = resolve_data_movement_delete_marker_resume_result(Ok(Some(source.clone())), &source, 0, 1) + .expect("equivalent delete marker target should be evaluated"); + + assert!(should_resume); + } + + #[test] + fn data_movement_delete_marker_resume_rejects_source_pool_target() { + let source = ObjectInfo { + version_id: Some(Uuid::nil()), + delete_marker: true, + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }; + + let should_resume = resolve_data_movement_delete_marker_resume_result(Ok(Some(source.clone())), &source, 0, 0) + .expect("source-pool target should be rejected before target lookup"); + + assert!(!should_resume); + } + + #[test] + fn data_movement_resume_target_prefers_selected_non_source_pool() { + let target_pool_idx = resolve_data_movement_resume_target_pool(2, Some(3), 1); + assert_eq!(target_pool_idx, 2); + } + + #[test] + fn data_movement_resume_target_uses_resolved_non_source_pool_when_selected_is_source() { + let target_pool_idx = resolve_data_movement_resume_target_pool(1, Some(3), 1); + assert_eq!(target_pool_idx, 3); + } + + #[test] + fn data_movement_resume_target_keeps_source_when_no_other_pool_is_available() { + let target_pool_idx = resolve_data_movement_resume_target_pool(1, None, 1); + assert_eq!(target_pool_idx, 1); + } + + #[test] + fn data_movement_delete_marker_resume_propagates_target_lookup_error() { + let source = ObjectInfo { + delete_marker: true, + ..Default::default() + }; + let result = resolve_data_movement_delete_marker_resume_result(Err(Error::SlowDown), &source, 0, 1); + + assert!(matches!(result, Err(Error::SlowDown))); + } + + #[test] + fn equivalent_data_movement_tiered_object_accepts_matching_transition_metadata() { + let version_id = Uuid::nil(); + let transition_version_id = Uuid::new_v4(); + let mod_time = OffsetDateTime::UNIX_EPOCH; + let source = FileInfo { + version_id: Some(version_id), + size: 1024, + mod_time: Some(mod_time), + checksum: Some(Bytes::from_static(b"checksum")), + transition_status: TRANSITION_COMPLETE.to_string(), + transitioned_objname: "remote/object".to_string(), + transition_tier: "WARM".to_string(), + transition_version_id: Some(transition_version_id), + metadata: HashMap::from([("etag".to_string(), "etag-value".to_string())]), + ..Default::default() + }; + let target = ObjectInfo { + version_id: Some(version_id), + size: 1024, + mod_time: Some(mod_time), + checksum: Some(Bytes::from_static(b"checksum")), + etag: Some("etag-value".to_string()), + transitioned_object: TransitionedObject { + name: "remote/object".to_string(), + version_id: transition_version_id.to_string(), + tier: "WARM".to_string(), + status: TRANSITION_COMPLETE.to_string(), + ..Default::default() + }, + ..Default::default() + }; + + assert!(is_equivalent_data_movement_tiered_object(&source, &target)); + } + + #[test] + fn equivalent_data_movement_tiered_object_rejects_transition_mismatch() { + let source = FileInfo { + version_id: Some(Uuid::nil()), + size: 1024, + transition_status: TRANSITION_COMPLETE.to_string(), + transitioned_objname: "remote/source".to_string(), + transition_tier: "WARM".to_string(), + ..Default::default() + }; + let target = ObjectInfo { + version_id: source.version_id, + size: 1024, + transitioned_object: TransitionedObject { + name: "remote/target".to_string(), + tier: "WARM".to_string(), + status: TRANSITION_COMPLETE.to_string(), + ..Default::default() + }, + ..Default::default() + }; + + assert!(!is_equivalent_data_movement_tiered_object(&source, &target)); + } + + #[test] + fn data_movement_tiered_resume_accepts_equivalent_target() { + let version_id = Uuid::nil(); + let transition_version_id = Uuid::new_v4(); + let source = FileInfo { + version_id: Some(version_id), + size: 1024, + transition_status: TRANSITION_COMPLETE.to_string(), + transitioned_objname: "remote/object".to_string(), + transition_tier: "WARM".to_string(), + transition_version_id: Some(transition_version_id), + metadata: HashMap::from([("etag".to_string(), "etag-value".to_string())]), + ..Default::default() + }; + let target = ObjectInfo { + version_id: Some(version_id), + size: 1024, + etag: Some("etag-value".to_string()), + transitioned_object: TransitionedObject { + name: "remote/object".to_string(), + version_id: transition_version_id.to_string(), + tier: "WARM".to_string(), + status: TRANSITION_COMPLETE.to_string(), + ..Default::default() + }, + ..Default::default() + }; + + let should_resume = resolve_data_movement_tiered_resume_result(Ok(Some(target)), &source, 0, 1) + .expect("equivalent tiered target should be evaluated"); + + assert!(should_resume); + } + + #[test] + fn data_movement_tiered_resume_rejects_source_pool_target() { + let version_id = Uuid::nil(); + let source = FileInfo { + version_id: Some(version_id), + size: 1024, + transition_status: TRANSITION_COMPLETE.to_string(), + transitioned_objname: "remote/object".to_string(), + transition_tier: "WARM".to_string(), + metadata: HashMap::from([("etag".to_string(), "etag-value".to_string())]), + ..Default::default() + }; + let target = ObjectInfo { + version_id: Some(version_id), + size: 1024, + etag: Some("etag-value".to_string()), + transitioned_object: TransitionedObject { + name: "remote/object".to_string(), + tier: "WARM".to_string(), + status: TRANSITION_COMPLETE.to_string(), + ..Default::default() + }, + ..Default::default() + }; + + let should_resume = resolve_data_movement_tiered_resume_result(Ok(Some(target)), &source, 0, 0) + .expect("source-pool target should be rejected before target lookup"); + + assert!(!should_resume); + } + + #[test] + fn data_movement_tiered_resume_rejects_missing_target() { + let source = FileInfo { + version_id: Some(Uuid::nil()), + size: 1024, + ..Default::default() + }; + + let should_resume = resolve_data_movement_tiered_resume_result(Ok(None), &source, 0, 1) + .expect("missing tiered target should be evaluated"); + + assert!(!should_resume); + } + #[test] fn latest_object_access_delete_marker_error_returns_none_for_live_object() { let info = ObjectInfo::default(); diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs index 64ff51a05..e1249ddfc 100644 --- a/crates/ecstore/src/store/rebalance.rs +++ b/crates/ecstore/src/store/rebalance.rs @@ -21,6 +21,11 @@ struct LatestObjectInfoCandidate { err: Option, } +struct RebalanceDeletePoolResult { + pool_idx: usize, + result: Result, +} + fn pool_lookup_not_found_error(bucket: &str, object: &str, opts: &ObjectOptions) -> Error { let object = decode_dir_object(object); @@ -39,6 +44,52 @@ fn resolve_rebalance_delete_from_all_pools_result(result: Result, bu result.map_err(|err| Error::other(format!("failed to delete rebalance source object {bucket}/{object}: {err}"))) } +fn is_ignorable_rebalance_delete_error(err: &Error) -> bool { + is_err_object_not_found(err) || is_err_version_not_found(err) +} + +fn rebalance_delete_pool_error(pool_idx: usize, bucket: &str, object: &str, err: Error) -> Error { + Error::other(format!("pool {pool_idx} delete failed for {bucket}/{object}: {err}")) +} + +fn resolve_rebalance_delete_from_all_pools_results( + results: Vec, + bucket: &str, + object: &str, +) -> Result { + let mut deleted = None; + let mut ignored_error = None; + + for pool_result in results { + let pool_idx = pool_result.pool_idx; + match pool_result.result { + Ok(info) => { + if deleted.is_none() { + deleted = Some(info); + } + } + Err(err) if is_ignorable_rebalance_delete_error(&err) => { + ignored_error = Some((pool_idx, err)); + } + Err(err) => { + return Err(rebalance_delete_pool_error(pool_idx, bucket, object, err)); + } + } + } + + if let Some(info) = deleted { + return Ok(info); + } + + if let Some((pool_idx, err)) = ignored_error { + return Err(rebalance_delete_pool_error(pool_idx, bucket, object, err)); + } + + Err(Error::other(format!( + "failed to delete rebalance source object {bucket}/{object}: no pools were attempted" + ))) +} + fn rebalance_disk_set_lookup_error(pool_idx: usize, set_idx: usize, pool_count: usize) -> Error { Error::other(format!( "failed to resolve rebalance disk set: pool index {pool_idx}, set index {set_idx}, pool count {pool_count}", @@ -220,6 +271,41 @@ impl ECStore { None } + pub(super) async fn get_available_pool_idx_excluding( + &self, + bucket: &str, + object: &str, + size: i64, + excluded_pool_idx: usize, + ) -> Option { + let mut server_pools = self.get_server_pools_available_space(bucket, object, size).await; + server_pools.filter_max_used(100 - (100_f64 * DISK_RESERVE_FRACTION) as u64); + + if let Some(pool) = server_pools.0.get_mut(excluded_pool_idx) { + pool.available = 0; + } + + let total = server_pools.total_available(); + if total == 0 { + return None; + } + + let mut rng = rand::rng(); + let random_u64: u64 = rng.random_range(0..total); + + let choose = random_u64 % total; + let mut at_total = 0; + + for pool in server_pools.iter() { + at_total += pool.available; + if at_total > choose && pool.available > 0 { + return Some(pool.index); + } + } + + None + } + async fn get_server_pools_available_space(&self, bucket: &str, object: &str, size: i64) -> ServerPoolsAvailableSpace { let mut n_sets = vec![0; self.pools.len()]; let mut infos = vec![Vec::new(); self.pools.len()]; @@ -507,38 +593,34 @@ impl ECStore { opts: &ObjectOptions, errs: Vec, ) -> Result { - let mut objs = Vec::new(); - let mut derrs = Vec::new(); + let mut results = Vec::with_capacity(errs.len()); for pe in errs.iter() { if let Some(err) = &pe.err && err == &StorageError::ErasureWriteQuorum { - objs.push(None); - derrs.push(Some(StorageError::ErasureWriteQuorum)); + if let Some(idx) = pe.index { + results.push(RebalanceDeletePoolResult { + pool_idx: idx, + result: Err(StorageError::ErasureWriteQuorum), + }); + } continue; } if let Some(idx) = pe.index { - match self.pools[idx].delete_object(bucket, object, opts.clone()).await { - Ok(res) => { - objs.push(Some(res)); - - derrs.push(None); - } - Err(err) => { - objs.push(None); - derrs.push(Some(err)); - } - } + results.push(RebalanceDeletePoolResult { + pool_idx: idx, + result: self.pools[idx].delete_object(bucket, object, opts.clone()).await, + }); } } - if let Some(e) = &derrs[0] { - return resolve_rebalance_delete_from_all_pools_result(Err(e.clone()), bucket, object); - } - - resolve_rebalance_delete_from_all_pools_result(Ok(objs[0].as_ref().unwrap().clone()), bucket, object) + resolve_rebalance_delete_from_all_pools_result( + resolve_rebalance_delete_from_all_pools_results(results, bucket, object), + bucket, + object, + ) } pub async fn reload_pool_meta(&self) -> Result<()> { @@ -902,6 +984,139 @@ mod tests { assert!(rendered.contains(&Error::SlowDown.to_string()), "{rendered}"); } + #[test] + fn resolve_rebalance_delete_from_all_pools_results_fails_on_later_pool_error() { + let err = resolve_rebalance_delete_from_all_pools_results( + vec![ + RebalanceDeletePoolResult { + pool_idx: 0, + result: Ok(ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + ..Default::default() + }), + }, + RebalanceDeletePoolResult { + pool_idx: 1, + result: Err(Error::SlowDown), + }, + ], + "bucket", + "object", + ) + .expect_err("non-ignorable errors from later pools must not be hidden"); + let rendered = err.to_string(); + + assert!(rendered.contains("pool 1 delete failed for bucket/object"), "{rendered}"); + assert!(rendered.contains(&Error::SlowDown.to_string()), "{rendered}"); + } + + #[test] + fn resolve_rebalance_delete_from_all_pools_results_ignores_later_not_found_after_success() { + let info = ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + ..Default::default() + }; + + let resolved = resolve_rebalance_delete_from_all_pools_results( + vec![ + RebalanceDeletePoolResult { + pool_idx: 0, + result: Ok(info.clone()), + }, + RebalanceDeletePoolResult { + pool_idx: 1, + result: Err(Error::ObjectNotFound("bucket".to_string(), "object".to_string())), + }, + ], + "bucket", + "object", + ) + .expect("not-found errors from other pools should be ignored when a delete succeeds"); + + assert_eq!(resolved.bucket, info.bucket); + assert_eq!(resolved.name, info.name); + } + + #[test] + fn resolve_rebalance_delete_from_all_pools_results_accepts_success_after_not_found() { + let info = ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + ..Default::default() + }; + + let resolved = resolve_rebalance_delete_from_all_pools_results( + vec![ + RebalanceDeletePoolResult { + pool_idx: 0, + result: Err(Error::ObjectNotFound("bucket".to_string(), "object".to_string())), + }, + RebalanceDeletePoolResult { + pool_idx: 1, + result: Ok(info.clone()), + }, + ], + "bucket", + "object", + ) + .expect("a successful delete should pass even when an earlier pool reports not-found"); + + assert_eq!(resolved.bucket, info.bucket); + assert_eq!(resolved.name, info.name); + } + + #[test] + fn resolve_rebalance_delete_from_all_pools_results_fails_when_all_results_are_ignored_errors() { + let err = resolve_rebalance_delete_from_all_pools_results( + vec![ + RebalanceDeletePoolResult { + pool_idx: 0, + result: Err(Error::ObjectNotFound("bucket".to_string(), "object".to_string())), + }, + RebalanceDeletePoolResult { + pool_idx: 1, + result: Err(Error::VersionNotFound("bucket".to_string(), "object".to_string(), "vid-1".to_string())), + }, + ], + "bucket", + "object", + ) + .expect_err("all ignored errors without any successful delete should still fail"); + let rendered = err.to_string(); + + assert!(rendered.contains("pool 1 delete failed for bucket/object"), "{rendered}"); + assert!(rendered.contains("Version not found"), "{rendered}"); + } + + #[test] + fn resolve_rebalance_delete_from_all_pools_results_fails_on_write_quorum_even_with_success() { + let err = resolve_rebalance_delete_from_all_pools_results( + vec![ + RebalanceDeletePoolResult { + pool_idx: 0, + result: Ok(ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + ..Default::default() + }), + }, + RebalanceDeletePoolResult { + pool_idx: 1, + result: Err(Error::ErasureWriteQuorum), + }, + ], + "bucket", + "object", + ) + .expect_err("write quorum failures must fail the aggregate delete"); + let rendered = err.to_string(); + + assert!(rendered.contains("pool 1 delete failed for bucket/object"), "{rendered}"); + assert!(rendered.contains(&Error::ErasureWriteQuorum.to_string()), "{rendered}"); + } + #[test] fn rebalance_disk_set_lookup_error_formats_pool_and_set_context() { let err = rebalance_disk_set_lookup_error(2, 7, 3); diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index dfed36381..8752415e9 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -147,6 +147,13 @@ fn should_yield_after_object(object_count: u64, yield_every: u64) -> bool { yield_every > 0 && object_count.is_multiple_of(yield_every) } +const SCANNER_FAILED_OBJECT_LOG_INITIAL_LIMIT: usize = 16; +const SCANNER_FAILED_OBJECT_LOG_EVERY: usize = 1024; + +fn should_log_failed_object(failed_objects: usize) -> bool { + failed_objects <= SCANNER_FAILED_OBJECT_LOG_INITIAL_LIMIT || failed_objects.is_multiple_of(SCANNER_FAILED_OBJECT_LOG_EVERY) +} + fn record_scanner_ilm_action_if_queued(metrics: &Metrics, count: u64, queued: bool) -> bool { if queued { metrics.record_scanner_ilm_action(count); @@ -1286,8 +1293,12 @@ impl FolderScanner { into.failed_objects += 1; self.record_failed(&item.path); - // Only log non-skip errors to avoid noise - warn!("scan_folder: failed to get size for item {}: {}", item.path, e); + if should_log_failed_object(into.failed_objects) { + warn!( + failed_objects = into.failed_objects, + "scan_folder: failed to get size for item {}: {}", item.path, e + ); + } } timer.sleep().await; @@ -2975,4 +2986,14 @@ mod tests { assert!(result.is_ok(), "expected symlinked child directory to be ignored"); assert_eq!(into.failed_objects, 0, "expected ignored symlink not to count as a failed object"); } + + #[test] + fn test_should_log_failed_object_samples_after_initial_limit() { + assert!(should_log_failed_object(1)); + assert!(should_log_failed_object(SCANNER_FAILED_OBJECT_LOG_INITIAL_LIMIT)); + assert!(!should_log_failed_object(SCANNER_FAILED_OBJECT_LOG_INITIAL_LIMIT + 1)); + assert!(should_log_failed_object(SCANNER_FAILED_OBJECT_LOG_EVERY)); + assert!(!should_log_failed_object(SCANNER_FAILED_OBJECT_LOG_EVERY + 1)); + assert!(should_log_failed_object(SCANNER_FAILED_OBJECT_LOG_EVERY * 2)); + } } diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 2666179e7..9f5142fb6 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -945,13 +945,11 @@ impl ScannerIODisk for Disk { let data = match self.read_metadata(&item.bucket, &item.object_path()).await { Ok(data) => data, Err(e) => { - warn!( - "Failed to read metadata: {e}, bucket={}, object_path={}", + return Err(StorageError::other(format!( + "failed to read metadata: {e}, bucket={}, object_path={}", &item.bucket, &item.object_path() - ); - - return Err(StorageError::other("failed to read metadata".to_string())); + ))); } };