diff --git a/crates/heal/src/heal/task/heal_bucket.rs b/crates/heal/src/heal/task/heal_bucket.rs index cb8d230b4..5795a743d 100644 --- a/crates/heal/src/heal/task/heal_bucket.rs +++ b/crates/heal/src/heal/task/heal_bucket.rs @@ -14,8 +14,107 @@ /// bucket/cluster/prefix heal: the recursive bucket-objects sweep and the erasure-set usage baseline use super::*; use crate::heal::progress::{add_bytes, increment_counter, stable_generation}; +use crate::heal::storage::HealListItem; use crate::heal::utils::format_set_disk_id; +const MAX_DEFERRED_OBJECTS: usize = 256; +const MAX_DEFERRED_BYTES: usize = 256 * 1024; +const MAX_DEFERRED_FORWARD_PAGES: u64 = 2; +const MAX_DEFERRED_AGE: Duration = Duration::from_secs(30); + +struct DeferredObject { + name: String, + version_id: Option, + attempt: u32, + page: u64, + first_failure: Option, + due: tokio::time::Instant, +} + +impl DeferredObject { + fn new(item: HealListItem, page: u64) -> Self { + Self { + name: item.name, + version_id: item.version_id, + attempt: 0, + page, + first_failure: None, + due: tokio::time::Instant::now(), + } + } + + fn payload_bytes(&self) -> usize { + self.name + .capacity() + .saturating_add(self.version_id.as_ref().map_or(0, String::capacity)) + } + + fn expired(&self) -> bool { + self.first_failure.is_some_and(|first| first.elapsed() >= MAX_DEFERRED_AGE) + } + + fn defer(&mut self, delay: Duration) { + let now = tokio::time::Instant::now(); + let first = *self.first_failure.get_or_insert(now); + self.attempt += 1; + self.due = (now + delay).min(first + MAX_DEFERRED_AGE); + } +} + +// Only failed identities are retained. The current listing page remains owned +// by the caller; capacity pressure stops fetching, never discards that page. +struct DeferredWindow { + objects: VecDeque, + bytes: usize, +} + +impl Default for DeferredWindow { + fn default() -> Self { + Self { + objects: VecDeque::new(), + // Charge every possible slot up front, including spare capacity. + bytes: MAX_DEFERRED_OBJECTS * size_of::(), + } + } +} + +impl DeferredWindow { + fn push(&mut self, item: DeferredObject) -> std::result::Result<(), DeferredObject> { + let bytes = item.payload_bytes(); + if self.objects.len() >= MAX_DEFERRED_OBJECTS || bytes > MAX_DEFERRED_BYTES.saturating_sub(self.bytes) { + return Err(item); + } + self.bytes += bytes; + self.objects.push_back(item); + Ok(()) + } + + fn pop_due(&mut self) -> Option { + let now = tokio::time::Instant::now(); + let index = self.objects.iter().position(|item| item.due <= now)?; + let item = self.objects.remove(index)?; + self.bytes -= item.payload_bytes(); + Some(item) + } + + fn next_due(&self) -> Option { + self.objects.iter().map(|item| item.due).min() + } + + fn can_advance(&self, page: u64) -> bool { + self.objects.len() < MAX_DEFERRED_OBJECTS + && self.bytes < MAX_DEFERRED_BYTES + && self + .objects + .iter() + .all(|item| page.saturating_sub(item.page) < MAX_DEFERRED_FORWARD_PAGES) + } +} + +#[cfg(test)] +#[path = "tests/deferred_retry_window.rs"] +mod deferred_retry_window; + fn unavailable_recreate_error(result: &HealResultItem, opts: &HealOpts) -> Option { if opts.dry_run || !opts.recreate { return None; @@ -305,83 +404,122 @@ impl HealTask { for (set_disk_id, heal_opts) in listing_scopes { let mut continuation_token: Option = None; - loop { - self.check_control_flags().await?; - let mut listing_attempt = 0; - let (objects, next_token, is_truncated) = loop { + let mut deferred = DeferredWindow::default(); + let mut inline_retry: Option = None; + let mut page_number = 0_u64; + let mut aborted_progress_unknown = false; + let mut pending = Vec::::new().into_iter(); + let mut listing_finished = false; + let mut listing_attempt = 0; + let mut listing_due = tokio::time::Instant::now(); + let scope_result: Result<()> = async { + loop { + self.check_control_flags().await?; + if listing_finished && pending.as_slice().is_empty() && deferred.objects.is_empty() && inline_retry.is_none() + { + break; + } self.pace_mainline().await?; - let page = if let Some(set_disk_id) = set_disk_id.as_deref() { - self.await_with_control(self.storage.list_versions_for_heal_page_disk_walk( - set_disk_id, - bucket, - prefix, - continuation_token.as_deref(), - false, - )) - .await - } else { - self.await_with_control(self.storage.list_objects_for_heal_page( - bucket, - prefix, - continuation_token.as_deref(), - false, - )) - .await - }; - match page { - Ok(page) => break page, - Err(error @ (Error::TaskCancelled | Error::TaskTimeout)) => return Err(error), - Err(error) => { - self.outcome.write().await.attempt_failed(); - if error.is_recoverable_heal() && listing_attempt < MAX_BUCKET_OBJECT_HEAL_RETRIES { - listing_attempt += 1; + // Listing and object retries share this safe boundary. A + // failed listing never hides an already-due object retry. + let item = deferred.pop_due().or_else(|| { + if inline_retry + .as_ref() + .is_some_and(|item| item.due <= tokio::time::Instant::now()) + { + inline_retry.take() + } else if inline_retry.is_none() { + pending.next().map(|item| DeferredObject::new(item, page_number)) + } else { + None + } + }); + let Some(mut item) = item else { + let can_list = !listing_finished && inline_retry.is_none() && deferred.can_advance(page_number); + if can_list && listing_due <= tokio::time::Instant::now() { + let page = if let Some(set_disk_id) = set_disk_id.as_deref() { + self.await_with_control(self.storage.list_versions_for_heal_page_disk_walk( + set_disk_id, + bucket, + prefix, + continuation_token.as_deref(), + false, + )) + .await + } else { + self.await_with_control(self.storage.list_objects_for_heal_page( + bucket, + prefix, + continuation_token.as_deref(), + false, + )) + .await + }; + match page { + Ok((objects, next_token, is_truncated)) => { + page_number = page_number.saturating_add(1); + continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?; + listing_finished = continuation_token.is_none(); + listing_attempt = 0; + listing_due = tokio::time::Instant::now(); + pending = objects.into_iter(); + } + Err(error @ (Error::TaskCancelled | Error::TaskTimeout)) => return Err(error), + Err(error) => { + self.outcome.write().await.attempt_failed(); + if error.is_recoverable_heal() && listing_attempt < MAX_BUCKET_OBJECT_HEAL_RETRIES { + listing_attempt += 1; + listing_due = + tokio::time::Instant::now() + self.bucket_object_retry_delay(listing_attempt); + continue; + } + self.outcome.write().await.mark_untraversable(); + return Err(Error::HealListingFailed { + bucket: bucket.to_string(), + source: Box::new(error), + }); + } + } + continue; + } else { + let due = deferred + .next_due() + .into_iter() + .chain(inline_retry.as_ref().map(|item| item.due)) + .chain(can_list.then_some(listing_due)) + .min(); + if let Some(due) = due { self.await_with_control(async { - tokio::time::sleep(self.bucket_object_retry_delay(listing_attempt)).await; + tokio::time::sleep_until(due).await; Ok(()) }) .await?; - continue; } - self.outcome.write().await.mark_untraversable(); - return Err(Error::HealListingFailed { - bucket: bucket.to_string(), - source: Box::new(error), - }); } + continue; + }; + let retry_attempt = item.attempt; + let mut telemetry_unknown = false; + let object = item.name.as_str(); + let identity = + self.outcome_identity(bucket, object, item.version_id.as_deref(), heal_opts.pool, heal_opts.set); + let mut disposition = if heal_opts.dry_run { + HealObjectDisposition::DryRunObserved + } else { + HealObjectDisposition::Unknown + }; + let mut detail = None; + { + let mut progress = self.progress.write().await; + progress.set_current_object(Some(format!("{bucket}/{object}"))); } - }; - let mut pending = objects; - let mut retry_attempt = 0_u32; - while !pending.is_empty() { - if retry_attempt > 0 { - self.await_with_control(async { - tokio::time::sleep(self.bucket_object_retry_delay(retry_attempt)).await; - Ok(()) - }) - .await?; - } - let mut retry = Vec::with_capacity(pending.len()); - for item in pending { - self.check_control_flags().await?; - self.pace_mainline().await?; - let mut telemetry_unknown = false; - let object = item.name.as_str(); - let identity = - self.outcome_identity(bucket, object, item.version_id.as_deref(), heal_opts.pool, heal_opts.set); - let mut disposition = if heal_opts.dry_run { - HealObjectDisposition::DryRunObserved - } else { - HealObjectDisposition::Unknown - }; - let mut detail = None; - { - let mut progress = self.progress.write().await; - progress.set_current_object(Some(format!("{bucket}/{object}"))); - } - - let mut terminal_outcome = true; - let error = match self + let mut terminal_outcome = true; + let age_exhausted = item.expired(); + let error = if age_exhausted { + Some(Error::other("heal object retry age exhausted")) + } else { + match self .await_with_control( self.storage .heal_object(bucket, object, item.version_id.as_deref(), &heal_opts), @@ -414,152 +552,188 @@ impl HealTask { None } Ok((_, Some(err))) | Err(err) => Some(err), - }; + } + }; - if let Some(err) = error { - match err { - Error::TaskCancelled | Error::TaskTimeout => { - let disposition = if matches!(err, Error::TaskCancelled) { - HealObjectDisposition::Cancelled - } else { - HealObjectDisposition::Deferred { - reason: HealDeferredReason::Deadline, - retry_not_before: None, - } - }; - self.outcome.write().await.record(HealObjectOutcome { - identity, - disposition, - detail: None, - }); - return Err(err); - } - _ => self.outcome.write().await.attempt_failed(), - } - detail = Some(err.to_string()); - if Self::is_dangling_delete_grace_error(&err) { - disposition = HealObjectDisposition::Deferred { - reason: HealDeferredReason::DanglingDeleteGrace, - retry_not_before: None, - }; - telemetry_unknown |= !increment_counter(&mut skipped); - warn!( - target: "rustfs::heal::task", - event = EVENT_HEAL_BUCKET_RESULT, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_TASK, - task_id = %self.id, - bucket, - object, - result = "dangling_delete_grace_skip", - error = %err, - "Heal bucket object dangling cleanup deferred by grace window" - ); - } else if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) { - disposition = HealObjectDisposition::Deferred { - reason: HealDeferredReason::TransientUsageCache, - retry_not_before: None, - }; - telemetry_unknown |= !increment_counter(&mut skipped); - warn!( - target: "rustfs::heal::task", - event = EVENT_HEAL_BUCKET_RESULT, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_TASK, - task_id = %self.id, - bucket, - object, - result = "transient_skip", - error = %err, - "Heal bucket object repair skipped due to transient metadata error" - ); - } else if err.is_recoverable_heal() && retry_attempt < MAX_BUCKET_OBJECT_HEAL_RETRIES { - terminal_outcome = false; - debug!( - target: "rustfs::heal::task", - event = EVENT_HEAL_BUCKET_RESULT, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_TASK, - task_id = %self.id, - bucket, - object, - retry_attempt = retry_attempt.saturating_add(1), - error = %err, - result = "object_retry_scheduled", - "Heal bucket object retry scheduled" - ); - retry.push(item); - } else { - disposition = HealObjectDisposition::Failed(if err.is_recoverable_heal() { - HealFailureClass::RetryExhausted + if let Some(err) = error { + match err { + Error::TaskCancelled | Error::TaskTimeout => { + let disposition = if matches!(err, Error::TaskCancelled) { + HealObjectDisposition::Cancelled } else { - HealFailureClass::Permanent + HealObjectDisposition::Deferred { + reason: HealDeferredReason::Deadline, + retry_not_before: None, + } + }; + self.outcome.write().await.record(HealObjectOutcome { + identity, + disposition, + detail: None, }); - telemetry_unknown |= !increment_counter(&mut failed); - if err.is_recoverable_heal() { - retryable_failed = retryable_failed.saturating_add(1); - } else { - permanent_failed = permanent_failed.saturating_add(1); - } - first_failed_object.get_or_insert_with(|| object.to_string()); - first_error.get_or_insert_with(|| err.to_string()); - if take_failure_log_sample(&mut failure_samples_logged) { - warn!( - target: "rustfs::heal::task", - event = EVENT_HEAL_BUCKET_RESULT, - component = LOG_COMPONENT_HEAL, - subsystem = LOG_SUBSYSTEM_TASK, - task_id = %self.id, - bucket, - object, - retry_attempt, - error = %err, - result = "object_failed", - "Heal bucket object repair failed" - ); - } + aborted_progress_unknown |= !increment_counter(&mut scanned); + aborted_progress_unknown |= !increment_counter(&mut skipped); + return Err(err); } + _ if !age_exhausted => self.outcome.write().await.attempt_failed(), + _ => {} } - - if terminal_outcome { - telemetry_unknown |= !increment_counter(&mut scanned); - } - - if !terminal_outcome { - continue; - } - - self.outcome.write().await.record(HealObjectOutcome { - identity, - disposition, - detail, - }); - - let mut progress = self.progress.write().await; - progress.update_object_progress( - previous_progress.objects_scanned.saturating_add(scanned), - previous_progress.objects_healed.saturating_add(healed), - previous_progress.objects_failed.saturating_add(failed), - previous_progress.skipped_objects.saturating_add(skipped), - previous_progress.bytes_processed.saturating_add(bytes), - ); - if telemetry_unknown { - progress.mark_unknown(); + detail = Some(err.to_string()); + if Self::is_dangling_delete_grace_error(&err) { + disposition = HealObjectDisposition::Deferred { + reason: HealDeferredReason::DanglingDeleteGrace, + retry_not_before: None, + }; + telemetry_unknown |= !increment_counter(&mut skipped); + warn!( + target: "rustfs::heal::task", + event = EVENT_HEAL_BUCKET_RESULT, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_TASK, + task_id = %self.id, + bucket, + object, + result = "dangling_delete_grace_skip", + error = %err, + "Heal bucket object dangling cleanup deferred by grace window" + ); + } else if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) { + disposition = HealObjectDisposition::Deferred { + reason: HealDeferredReason::TransientUsageCache, + retry_not_before: None, + }; + telemetry_unknown |= !increment_counter(&mut skipped); + warn!( + target: "rustfs::heal::task", + event = EVENT_HEAL_BUCKET_RESULT, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_TASK, + task_id = %self.id, + bucket, + object, + result = "transient_skip", + error = %err, + "Heal bucket object repair skipped due to transient metadata error" + ); + } else if !age_exhausted && err.is_recoverable_heal() && retry_attempt < MAX_BUCKET_OBJECT_HEAL_RETRIES { + terminal_outcome = false; + debug!( + target: "rustfs::heal::task", + event = EVENT_HEAL_BUCKET_RESULT, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_TASK, + task_id = %self.id, + bucket, + object, + retry_attempt = retry_attempt.saturating_add(1), + error = %err, + result = "object_retry_scheduled", + "Heal bucket object retry scheduled" + ); + item.defer(self.bucket_object_retry_delay(retry_attempt + 1)); + if let Err(item) = deferred.push(item) { + inline_retry = Some(item); + } + } else { + disposition = HealObjectDisposition::Failed(if age_exhausted || err.is_recoverable_heal() { + HealFailureClass::RetryExhausted + } else { + HealFailureClass::Permanent + }); + telemetry_unknown |= !increment_counter(&mut failed); + if age_exhausted || err.is_recoverable_heal() { + retryable_failed = retryable_failed.saturating_add(1); + } else { + permanent_failed = permanent_failed.saturating_add(1); + } + first_failed_object.get_or_insert_with(|| object.to_string()); + first_error.get_or_insert_with(|| err.to_string()); + if take_failure_log_sample(&mut failure_samples_logged) { + warn!( + target: "rustfs::heal::task", + event = EVENT_HEAL_BUCKET_RESULT, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_TASK, + task_id = %self.id, + bucket, + object, + retry_attempt, + error = %err, + result = "object_failed", + "Heal bucket object repair failed" + ); + } } } - pending = retry; - retry_attempt = retry_attempt.saturating_add(1); - } - if !is_truncated { - break; - } + if terminal_outcome { + telemetry_unknown |= !increment_counter(&mut scanned); + } - continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?; - if continuation_token.is_none() { - // Truncated without a continuation token is a compatibility EOF. - break; + if !terminal_outcome { + continue; + } + + self.outcome.write().await.record(HealObjectOutcome { + identity, + disposition, + detail, + }); + + let mut progress = self.progress.write().await; + progress.update_object_progress( + previous_progress.objects_scanned.saturating_add(scanned), + previous_progress.objects_healed.saturating_add(healed), + previous_progress.objects_failed.saturating_add(failed), + previous_progress.skipped_objects.saturating_add(skipped), + previous_progress.bytes_processed.saturating_add(bytes), + ); + if telemetry_unknown { + progress.mark_unknown(); + } } + Ok(()) + } + .await; + if let Err(error) = scope_result { + let disposition = match error { + Error::TaskCancelled => HealObjectDisposition::Cancelled, + Error::TaskTimeout => HealObjectDisposition::Deferred { + reason: HealDeferredReason::Deadline, + retry_not_before: None, + }, + _ => HealObjectDisposition::Unknown, + }; + // Only attempted identities have terminal outcomes. Unstarted + // page tails remain unprocessed under the task's partial coverage. + // No detached sleepers survive abort. + for item in deferred.objects.into_iter().chain(inline_retry) { + self.outcome.write().await.record(HealObjectOutcome { + identity: self.outcome_identity( + bucket, + &item.name, + item.version_id.as_deref(), + heal_opts.pool, + heal_opts.set, + ), + disposition: disposition.clone(), + detail: None, + }); + aborted_progress_unknown |= !increment_counter(&mut scanned); + aborted_progress_unknown |= !increment_counter(&mut skipped); + } + let mut progress = self.progress.write().await; + progress.update_object_progress( + previous_progress.objects_scanned.saturating_add(scanned), + previous_progress.objects_healed.saturating_add(healed), + previous_progress.objects_failed.saturating_add(failed), + previous_progress.skipped_objects.saturating_add(skipped), + previous_progress.bytes_processed.saturating_add(bytes), + ); + if aborted_progress_unknown { + progress.mark_unknown(); + } + return Err(error); } } diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 8cabf1214..788b356ac 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -15,6 +15,8 @@ use super::super::{DiskOption, DiskStore, Endpoint, new_disk}; use super::*; +mod deferred_retry; + mod canonical_outcome { use super::*; use crate::heal::outcome::{HealExecutionOutcome, HealTraversalCoverage}; @@ -939,6 +941,10 @@ async fn verified_recovery_keeps_state_when_marker_clear_fails() { #[derive(Default)] struct MockStorage { + retry_test_pages: Option>>, + retry_test_delays: HashMap, + retry_test_listing_delays: Mutex>, + retry_test_events: Mutex>, listed: Mutex, list_each_bucket: bool, fail_second_listing_page: bool, @@ -1109,6 +1115,7 @@ fn replacement_identity( } enum MockHealObjectOutcome { + RetryableLock, OkWithOtherError(&'static str), ErrOther(&'static str), DanglingGraceDeferred, @@ -1213,6 +1220,10 @@ impl HealStorageAPI for MockStorage { opts: &HealOpts, ) -> Result<(HealResultItem, Option)> { self.heal_object_calls.lock().unwrap().push(object.to_string()); + self.retry_test_events.lock().expect("events").push(format!("heal:{object}")); + if let Some(delay) = self.retry_test_delays.get(object) { + tokio::time::sleep(*delay).await; + } self.heal_object_version_ids .lock() .unwrap() @@ -1241,6 +1252,13 @@ impl HealStorageAPI for MockStorage { bucket.to_string(), object.to_string(), ))), + MockHealObjectOutcome::RetryableLock => Ok(( + HealResultItem::default(), + Some(Error::Storage(EcstoreError::Lock(rustfs_lock::LockError::AlreadyLocked { + resource: object.to_string(), + owner: "competing-writer".to_string(), + }))), + )), MockHealObjectOutcome::RetryableSlowDown => { Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::SlowDown)))) } @@ -1266,6 +1284,13 @@ impl HealStorageAPI for MockStorage { bucket.to_string(), object.to_string(), ))), + MockHealObjectOutcome::RetryableLock => Ok(( + HealResultItem::default(), + Some(Error::Storage(EcstoreError::Lock(rustfs_lock::LockError::AlreadyLocked { + resource: object.to_string(), + owner: "competing-writer".to_string(), + }))), + )), MockHealObjectOutcome::RetryableSlowDown => { Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::SlowDown)))) } @@ -1361,6 +1386,19 @@ impl HealStorageAPI for MockStorage { .lock() .expect("listing tokens") .push(continuation_token.map(ToOwned::to_owned)); + self.retry_test_events + .lock() + .expect("events") + .push(format!("list:{}", continuation_token.unwrap_or("first"))); + let delay = self.retry_test_listing_delays.lock().expect("listing delays").pop_front(); + if let Some(delay) = delay { + tokio::time::sleep(delay).await; + } + if let Some(pages) = &self.retry_test_pages { + let page = continuation_token.map_or(0, |token| token.parse::().expect("test page token")); + let next = (page + 1 < pages.len()).then(|| (page + 1).to_string()); + return Ok((pages[page].clone(), next.clone(), next.is_some())); + } if let Some(remaining) = self .recoverable_second_page_failures .lock() diff --git a/crates/heal/src/heal/task/tests/deferred_retry.rs b/crates/heal/src/heal/task/tests/deferred_retry.rs new file mode 100644 index 000000000..c50c73551 --- /dev/null +++ b/crates/heal/src/heal/task/tests/deferred_retry.rs @@ -0,0 +1,487 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::*; + +fn bucket_task(storage: Arc) -> HealTask { + HealTask::from_request( + HealRequest::new( + HealType::Bucket { + bucket: "bucket-a".to_string(), + }, + HealOptions { + recursive: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ), + storage, + ) +} + +fn pages_storage(pages: &[&[&str]]) -> MockStorage { + MockStorage { + retry_test_pages: Some( + pages + .iter() + .map(|page| page.iter().map(|name| heal_item(name)).collect()) + .collect(), + ), + ..Default::default() + } +} + +fn fail_once(storage: &MockStorage, name: &str) { + storage + .heal_object_outcomes + .lock() + .expect("outcomes") + .insert(name.to_string(), VecDeque::from([MockHealObjectOutcome::RetryableLock])); +} + +#[tokio::test(start_paused = true)] +async fn slow_listing_retry_services_due_object_then_age_before_next_listing() { + let storage = Arc::new(MockStorage { + recoverable_second_page_failures: Mutex::new(Some(1)), + retry_test_listing_delays: Mutex::new(VecDeque::from([Duration::ZERO, Duration::from_secs(29)])), + ..Default::default() + }); + storage.heal_object_outcomes.lock().expect("outcomes").insert( + "object-a".to_string(), + VecDeque::from([ + MockHealObjectOutcome::RetryableSlowDown, + MockHealObjectOutcome::RetryableSlowDown, + ]), + ); + let task = bucket_task(storage.clone()); + let execution = task.execute(); + tokio::pin!(execution); + assert!( + tokio::time::timeout(Duration::from_millis(30_500), &mut execution) + .await + .is_err() + ); + assert_eq!( + storage.retry_test_events.lock().expect("events").as_slice(), + ["list:first", "heal:object-a", "list:second", "heal:object-a"] + ); + let outcome = task.get_outcome().await; + assert_eq!(outcome.counters.processed, 1); + assert_eq!( + outcome.objects[0].disposition, + HealObjectDisposition::Failed(HealFailureClass::RetryExhausted) + ); + execution.await.expect_err("age exhausted object must remain a batch failure"); + assert_eq!( + storage.retry_test_events.lock().expect("events").as_slice(), + [ + "list:first", + "heal:object-a", + "list:second", + "heal:object-a", + "list:second", + "heal:object-b" + ] + ); + let outcome = task.get_outcome().await; + assert_eq!((outcome.counters.processed, outcome.counters.attempt_failures), (2, 3)); +} + +#[tokio::test(start_paused = true)] +async fn listing_return_after_age_expires_does_not_start_another_heal_attempt() { + let storage = Arc::new(MockStorage { + recoverable_second_page_failures: Mutex::new(Some(1)), + retry_test_listing_delays: Mutex::new(VecDeque::from([Duration::ZERO, Duration::from_secs(31)])), + ..Default::default() + }); + fail_once(&storage, "object-a"); + let task = bucket_task(storage.clone()); + let execution = task.execute(); + tokio::pin!(execution); + assert!( + tokio::time::timeout(Duration::from_millis(31_500), &mut execution) + .await + .is_err() + ); + assert_eq!( + storage.retry_test_events.lock().expect("events").as_slice(), + ["list:first", "heal:object-a", "list:second"] + ); + assert_eq!(task.get_outcome().await.counters.failed, 1); + execution.await.expect_err("age exhaustion remains a failure"); + assert_eq!( + storage.retry_test_events.lock().expect("events").as_slice(), + ["list:first", "heal:object-a", "list:second", "list:second", "heal:object-b"] + ); +} + +#[tokio::test(start_paused = true)] +async fn full_window_abort_accounts_inline_once_and_leaves_unstarted_tail_unprocessed() { + for cancel in [true, false] { + let names: Vec = (0..258).map(|index| format!("blocked-{index}")).collect(); + let mut page: Vec = names.iter().map(|name| heal_item(name)).collect(); + page[256].version_id = Some("inline-version".to_string()); + let storage = Arc::new(MockStorage { + retry_test_pages: Some(vec![page, vec![heal_item("healthy")]]), + ..Default::default() + }); + for name in &names { + fail_once(&storage, name); + } + let mut task = bucket_task(storage.clone()); + if !cancel { + task.options.timeout = Some(Duration::from_secs(1)); + } + let execution = task.execute(); + tokio::pin!(execution); + assert!( + tokio::time::timeout(Duration::from_millis(500), &mut execution) + .await + .is_err() + ); + assert_eq!(storage.heal_object_calls.lock().expect("calls").len(), 257); + assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 1); + if cancel { + task.cancel().await.expect("cancel"); + } + let result = execution.await; + assert!(matches!( + (&result, cancel), + (Err(Error::TaskCancelled), true) | (Err(Error::TaskTimeout), false) + )); + let outcome = task.get_outcome().await; + assert_eq!( + ( + outcome.counters.processed, + outcome.counters.skipped, + outcome.counters.failed, + outcome.counters.healed + ), + (257, 257, 0, 0) + ); + assert_eq!(outcome.coverage, crate::heal::outcome::HealTraversalCoverage::Partial); + assert_eq!( + outcome.execution, + crate::heal::outcome::HealExecutionOutcome::Aborted(if cancel { + HealAbortReason::Cancelled + } else { + HealAbortReason::Deadline + }) + ); + let inline: Vec<_> = outcome + .objects + .iter() + .filter(|item| item.identity.object == "blocked-256") + .collect(); + assert_eq!(inline.len(), 1); + assert_eq!(inline[0].identity.version_id.as_deref(), Some("inline-version")); + assert_eq!( + inline[0].disposition, + if cancel { + HealObjectDisposition::Cancelled + } else { + HealObjectDisposition::Deferred { + reason: HealDeferredReason::Deadline, + retry_not_before: None, + } + } + ); + let progress = task.get_progress().await; + assert_eq!( + ( + progress.objects_scanned, + progress.skipped_objects, + progress.objects_failed, + progress.objects_healed + ), + (257, 257, 0, 0) + ); + assert!( + !outcome + .objects + .iter() + .any(|item| item.identity.object == "blocked-257" || item.identity.object == "healthy") + ); + tokio::time::advance(Duration::from_secs(60)).await; + assert_eq!(storage.heal_object_calls.lock().expect("calls").len(), 257); + assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 1); + } +} + +#[tokio::test(start_paused = true)] +async fn repeated_slowdown_keeps_attempts_and_forward_pages_bounded() { + let storage = Arc::new(pages_storage(&[&["a"], &["b"], &["c"], &["d"]])); + for name in ["a", "b", "c", "d"] { + storage + .heal_object_outcomes + .lock() + .expect("outcomes") + .insert(name.to_string(), (0..4).map(|_| MockHealObjectOutcome::RetryableSlowDown).collect()); + } + let task = bucket_task(storage.clone()); + let execution = task.execute(); + tokio::pin!(execution); + assert!(tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err()); + assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 3); + execution.await.expect_err("all four objects exhaust retries"); + let outcome = task.get_outcome().await; + assert_eq!( + (outcome.counters.processed, outcome.counters.failed, outcome.counters.attempt_failures), + (4, 4, 16) + ); + assert_eq!(storage.heal_object_calls.lock().expect("calls").len(), 16); + for name in ["a", "b", "c", "d"] { + assert_eq!(outcome.objects.iter().filter(|item| item.identity.object == name).count(), 1); + } +} + +#[tokio::test(start_paused = true)] +async fn listing_retry_keeps_cursor_and_does_not_replay_successful_objects() { + let storage = Arc::new(MockStorage { + recoverable_second_page_failures: Mutex::new(Some(1)), + ..Default::default() + }); + fail_once(&storage, "object-a"); + let task = bucket_task(storage.clone()); + task.execute().await.expect("both retries complete"); + assert_eq!( + storage.listing_tokens.lock().expect("tokens").as_slice(), + [None, Some("second".to_string()), Some("second".to_string())] + ); + assert_eq!( + storage.heal_object_calls.lock().expect("calls").as_slice(), + ["object-a", "object-a", "object-b"] + ); + let outcome = task.get_outcome().await; + assert_eq!((outcome.counters.processed, outcome.counters.attempt_failures), (2, 2)); +} + +#[tokio::test(start_paused = true)] +async fn typed_lock_contention_allows_only_two_forward_pages() { + let storage = Arc::new(pages_storage(&[&["a"], &["b"], &["c"], &["d"]])); + fail_once(&storage, "a"); + let task = bucket_task(storage.clone()); + let execution = task.execute(); + tokio::pin!(execution); + assert!(tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err()); + assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b", "c"]); + assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 3); + execution.await.expect("all objects complete"); + assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b", "c", "a", "d"]); + assert_eq!(task.get_outcome().await.counters.processed, 4); +} + +#[tokio::test(start_paused = true)] +async fn due_retry_runs_before_next_object_in_a_slow_healthy_page() { + let mut storage = pages_storage(&[&["a"], &["b", "c"]]); + storage.retry_test_delays.insert("b".to_string(), Duration::from_secs(3)); + fail_once(&storage, "a"); + let storage = Arc::new(storage); + bucket_task(storage.clone()).execute().await.expect("all objects complete"); + assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b", "a", "c"]); +} + +#[tokio::test(start_paused = true)] +async fn expired_retry_is_terminal_without_an_extra_storage_attempt() { + let mut storage = pages_storage(&[&["a"], &["b"]]); + storage.retry_test_delays.insert("b".to_string(), Duration::from_secs(31)); + fail_once(&storage, "a"); + let storage = Arc::new(storage); + let task = bucket_task(storage.clone()); + task.execute().await.expect_err("aged pending responsibility is not success"); + assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b"]); + let outcome = task.get_outcome().await; + assert_eq!(outcome.counters.processed, 2); + assert_eq!(outcome.counters.attempt_failures, 1); + assert_eq!(outcome.counters.failed, 1); + assert_eq!( + outcome + .objects + .iter() + .find(|item| item.identity.object == "a") + .expect("a outcome") + .disposition, + HealObjectDisposition::Failed(HealFailureClass::RetryExhausted) + ); +} + +#[tokio::test(start_paused = true)] +async fn cancellation_drains_owned_retries_once() { + let storage = Arc::new(pages_storage(&[&["a"], &["b"]])); + fail_once(&storage, "a"); + let task = bucket_task(storage.clone()); + let execution = task.execute(); + tokio::pin!(execution); + assert!(tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err()); + task.cancel().await.expect("cancel"); + assert!(matches!(execution.await, Err(Error::TaskCancelled))); + tokio::time::advance(Duration::from_secs(60)).await; + assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b"]); + let outcome = task.get_outcome().await; + assert_eq!(outcome.counters.processed, 2); + assert_eq!(outcome.objects.iter().filter(|item| item.identity.object == "a").count(), 1); + assert_eq!( + outcome + .objects + .iter() + .find(|item| item.identity.object == "a") + .expect("a") + .disposition, + HealObjectDisposition::Cancelled + ); +} + +#[tokio::test(start_paused = true)] +async fn deadline_drains_owned_retries_without_false_completion() { + let storage = Arc::new(pages_storage(&[&["a"], &["b"]])); + fail_once(&storage, "a"); + let mut task = bucket_task(storage.clone()); + task.options.timeout = Some(Duration::from_secs(1)); + assert!(matches!(task.execute().await, Err(Error::TaskTimeout))); + let outcome = task.get_outcome().await; + assert_eq!(outcome.counters.processed, 2); + assert_eq!( + outcome + .objects + .iter() + .find(|item| item.identity.object == "a") + .expect("a") + .disposition, + HealObjectDisposition::Deferred { + reason: HealDeferredReason::Deadline, + retry_not_before: None + } + ); + tokio::time::advance(Duration::from_secs(60)).await; + assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["a", "b"]); +} + +#[tokio::test(start_paused = true)] +async fn full_window_backpressures_without_losing_the_current_page_tail() { + let names: Vec = (0..258).map(|index| format!("blocked-{index}")).collect(); + let mut storage = MockStorage { + retry_test_pages: Some(vec![names.iter().map(|name| heal_item(name)).collect(), vec![heal_item("healthy")]]), + ..Default::default() + }; + for name in &names { + fail_once(&storage, name); + } + // The last item has a version, proving the current-page tail is not rebuilt + // from names alone when the window fills. + storage.retry_test_pages.as_mut().expect("pages")[0][257].version_id = Some("version-tail".to_string()); + let storage = Arc::new(storage); + let task = bucket_task(storage.clone()); + let execution = task.execute(); + tokio::pin!(execution); + assert!(tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err()); + assert_eq!(storage.heal_object_calls.lock().expect("calls").len(), 257); + assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 1); + execution.await.expect("every owned item eventually completes"); + assert_eq!(task.get_outcome().await.counters.processed, 259); + let calls = storage.heal_object_calls.lock().expect("calls"); + for name in &names { + assert_eq!(calls.iter().filter(|called| *called == name).count(), 2); + } + let versions = storage.heal_object_version_ids.lock().expect("versions"); + for (name, version) in calls.iter().zip(versions.iter()) { + if name == "blocked-257" { + assert_eq!(version.as_deref(), Some("version-tail")); + } + } +} + +#[tokio::test(start_paused = true)] +async fn oversized_identity_stays_inline_without_losing_version() { + let name = "k".repeat(256 * 1024); + let mut item = heal_item(&name); + item.version_id = Some("v".repeat(1024)); + let storage = Arc::new(MockStorage { + retry_test_pages: Some(vec![vec![item], vec![heal_item("healthy")]]), + ..Default::default() + }); + fail_once(&storage, &name); + let task = bucket_task(storage.clone()); + let execution = task.execute(); + tokio::pin!(execution); + assert!(tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err()); + assert_eq!(storage.listing_tokens.lock().expect("tokens").len(), 1); + execution.await.expect("oversized identity retries inline"); + assert_eq!(task.get_outcome().await.counters.processed, 2); + let versions = storage.heal_object_version_ids.lock().expect("versions"); + assert_eq!(versions[0], versions[1]); + assert_eq!(versions[0].as_ref().expect("version").len(), 1024); +} + +#[tokio::test(start_paused = true)] +async fn terminal_listing_failure_keeps_deferred_identity_unknown() { + let storage = Arc::new(MockStorage { + fail_second_listing_page: true, + ..Default::default() + }); + fail_once(&storage, "object-a"); + let task = bucket_task(storage.clone()); + task.execute().await.expect_err("listing cannot continue"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.counters.processed, 1); + assert_eq!(outcome.objects[0].disposition, HealObjectDisposition::Unknown); + assert_eq!(outcome.coverage, crate::heal::outcome::HealTraversalCoverage::Partial); + assert_eq!(storage.heal_object_calls.lock().expect("calls").as_slice(), ["object-a"]); +} + +#[tokio::test(start_paused = true)] +async fn healthy_second_page_advances_before_first_retry_is_due() { + let storage = Arc::new(MockStorage { + recoverable_second_page_failures: Mutex::new(Some(0)), + ..Default::default() + }); + storage + .heal_object_outcomes + .lock() + .expect("outcomes") + .insert("object-a".to_string(), VecDeque::from([MockHealObjectOutcome::RetryableSlowDown])); + let task = HealTask::from_request( + HealRequest::new( + HealType::Bucket { + bucket: "bucket-a".to_string(), + }, + HealOptions { + recursive: true, + timeout: None, + ..Default::default() + }, + HealPriority::Normal, + ), + storage.clone(), + ); + let execution = task.execute(); + tokio::pin!(execution); + assert!( + tokio::time::timeout(Duration::from_secs(1), &mut execution).await.is_err(), + "the deferred first object must remain pending before its retry is due" + ); + assert_eq!( + storage.heal_object_calls.lock().expect("calls").as_slice(), + ["object-a", "object-b"], + "a retryable page head must not hold the healthy second page behind its backoff" + ); + execution.await.expect("retry eventually succeeds"); + let outcome = task.get_outcome().await; + assert_eq!(outcome.counters.processed, 2); + assert_eq!(outcome.counters.attempt_failures, 1); + assert_eq!( + storage.heal_object_calls.lock().expect("calls").as_slice(), + ["object-a", "object-b", "object-a"] + ); +} diff --git a/crates/heal/src/heal/task/tests/deferred_retry_window.rs b/crates/heal/src/heal/task/tests/deferred_retry_window.rs new file mode 100644 index 000000000..fe93aa1e0 --- /dev/null +++ b/crates/heal/src/heal/task/tests/deferred_retry_window.rs @@ -0,0 +1,76 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::*; + +fn item(name: String, version_id: Option) -> DeferredObject { + DeferredObject::new( + HealListItem { + name, + version_id, + mod_time_unix_nanos: None, + lifecycle_object_info: None, + is_delete_marker: false, + }, + 1, + ) +} + +#[tokio::test(start_paused = true)] +async fn count_cap_and_next_item_preserve_ownership() { + let mut window = DeferredWindow::default(); + for _ in 0..MAX_DEFERRED_OBJECTS { + assert!(window.push(item("key".to_string(), None)).is_ok()); + } + let rejected = window + .push(item("next".to_string(), Some("version".to_string()))) + .expect_err("count cap"); + assert_eq!(rejected.name, "next"); + assert_eq!(rejected.version_id.as_deref(), Some("version")); + assert_eq!(window.objects.len(), MAX_DEFERRED_OBJECTS); + assert!(window.bytes <= MAX_DEFERRED_BYTES); + assert!(!window.can_advance(1)); + assert!(window.pop_due().is_some()); + assert!(window.push(rejected).is_ok()); +} + +#[tokio::test(start_paused = true)] +async fn byte_cap_counts_key_version_and_reserved_slots() { + let mut window = DeferredWindow::default(); + let available = MAX_DEFERRED_BYTES - window.bytes; + let key = "k".repeat(available / 2); + let version = "v".repeat(available - key.capacity()); + assert_eq!(key.capacity() + version.capacity(), available); + assert!(window.push(item(key, Some(version))).is_ok()); + assert_eq!(window.bytes, MAX_DEFERRED_BYTES); + assert!(!window.can_advance(1)); + assert!(window.push(item("x".to_string(), None)).is_err()); + assert!(window.pop_due().is_some()); + assert_eq!(window.bytes, MAX_DEFERRED_OBJECTS * size_of::()); + assert!(window.push(item("x".to_string(), None)).is_ok()); +} + +#[tokio::test(start_paused = true)] +async fn retry_age_caps_due_time_and_is_not_reset_by_rescheduling() { + let mut entry = item("a".to_string(), None); + entry.defer(Duration::from_secs(2)); + let first = entry.first_failure.expect("first failure"); + tokio::time::advance(Duration::from_secs(29)).await; + entry.defer(Duration::from_secs(8)); + assert_eq!(entry.first_failure, Some(first)); + assert_eq!(entry.due, first + MAX_DEFERRED_AGE); + assert!(!entry.expired()); + tokio::time::advance(Duration::from_secs(1)).await; + assert!(entry.expired()); +}