From 6b1554003ebf8f2037ffb7da9c9b906527e758da Mon Sep 17 00:00:00 2001 From: cxymds Date: Wed, 7 Oct 2026 22:04:12 +0800 Subject: [PATCH] fix(heal): defer replacement retries until dangling grace expires (#8388) --- crates/heal/src/error.rs | 9 +- crates/heal/src/heal/erasure_healer.rs | 98 +++- .../resume_loop_tests/dangling_grace.rs | 473 ++++++++++++++++++ crates/heal/src/heal/manager.rs | 37 +- crates/heal/src/heal/manager/auto_scan.rs | 7 + crates/heal/src/heal/manager/scheduler.rs | 53 +- crates/heal/src/heal/manager/tests.rs | 225 +++++++++ .../heal/src/heal/manager/unclean_shutdown.rs | 15 + crates/heal/src/heal/resume.rs | 45 ++ crates/heal/src/heal/resume/checkpoint.rs | 41 +- crates/heal/src/heal/resume/replacement.rs | 17 + crates/heal/src/heal/resume/tests.rs | 290 +++++++++++ 12 files changed, 1292 insertions(+), 18 deletions(-) create mode 100644 crates/heal/src/heal/erasure_healer/resume_loop_tests/dangling_grace.rs diff --git a/crates/heal/src/error.rs b/crates/heal/src/error.rs index b00f0000b..56a6f23f5 100644 --- a/crates/heal/src/error.rs +++ b/crates/heal/src/error.rs @@ -75,6 +75,9 @@ pub enum Error { #[error("replacement recovery retry budget exhausted")] ReplacementRetryBudgetExhausted, + #[error("Dangling object cleanup deferred until UNIX time {retry_not_before}")] + DanglingDeleteDeferred { retry_not_before: u64 }, + #[error("stale_bucket_incarnation: bucket {bucket} no longer belongs to this heal admission ({expected:?})")] StaleBucketIncarnation { bucket: String, expected: Option }, @@ -123,7 +126,7 @@ impl Error { pub(crate) fn is_recoverable_heal(&self) -> bool { match self { Error::TaskCancelled | Error::TaskTimeout | Error::StaleBucketIncarnation { .. } => false, - Error::ReplacementTargetNotReady(_) => true, + Error::ReplacementTargetNotReady(_) | Error::DanglingDeleteDeferred { .. } => true, Error::TransientSkip { .. } => true, // Lock failures classify by LockError's own taxonomy: only the // fatal variants (ResourceNotFound / PermissionDenied / @@ -193,6 +196,9 @@ impl Error { } pub(crate) fn dangling_delete_retry_not_before(&self) -> Option { + if let Self::DanglingDeleteDeferred { retry_not_before } = self { + return std::time::UNIX_EPOCH.checked_add(std::time::Duration::from_secs(*retry_not_before)); + } let after = match self { Self::Storage(error) => error.dangling_delete_retry_after(), Self::Disk(error) => error.dangling_delete_retry_after(), @@ -204,6 +210,7 @@ impl Error { pub(crate) fn is_dangling_delete_grace(&self) -> bool { match self { + Error::DanglingDeleteDeferred { .. } => true, Error::Storage(err) => err.is_dangling_delete_grace(), Error::Disk(err) => err.is_dangling_delete_grace(), Error::Io(err) => DiskError::io_error_is_dangling_delete_grace(err), diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index d2d1b4c3f..32fee9ff6 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -30,7 +30,7 @@ use std::sync::{ Arc, atomic::{AtomicUsize, Ordering}, }; -use std::time::{Duration, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::{RwLock, Semaphore}; use tracing::{debug, error, warn}; @@ -44,6 +44,9 @@ enum HealObjectOutcome { /// contended lock). The object is skipped and retried on a later pass /// instead of being recorded as processed. Transient, + /// A storage safety window with a known deadline; it consumes no failure + /// retry when it is the only reason a replacement pass remains incomplete. + DanglingDeleteDeferred { retry_not_before: u64 }, /// A real heal failure that should be recorded as failed. Failed, } @@ -312,7 +315,21 @@ impl ErasureSetHealer { /// Most heal object failures wrap `Error::Storage(StorageError)`, while /// compatibility markers can also arrive through Disk/Io/task wrappers. fn classify_heal_object_error(err: &Error) -> HealObjectOutcome { + if let Error::DanglingDeleteDeferred { retry_not_before } = err { + return HealObjectOutcome::DanglingDeleteDeferred { + retry_not_before: *retry_not_before, + }; + } if err.is_dangling_delete_grace() { + if let Some(retry_not_before) = err + .dangling_delete_retry_not_before() + .and_then(|deadline| deadline.duration_since(UNIX_EPOCH).ok()) + .and_then(|deadline| deadline.as_secs().checked_add(1)) + { + // Storage reports whole remaining seconds. Round the absolute + // deadline up so subsecond grace boundaries cannot hot-loop. + return HealObjectOutcome::DanglingDeleteDeferred { retry_not_before }; + } return HealObjectOutcome::Transient; } @@ -622,7 +639,7 @@ impl ErasureSetHealer { let state = resume_manager.get_state().await; if self.admin_task.is_none() - && state.retry_count > 0 + && (state.retry_count > 0 || state.dangling_delete_retry_not_before.is_some()) && state.completed_buckets.is_empty() && state.resume_cursor.is_none() && state.processed_objects == 0 @@ -633,9 +650,9 @@ impl ErasureSetHealer { && state.skipped_ilm_expired == 0 && state.processed_bytes == 0 { - // schedule_retry persists the authoritative resume reset before - // resetting the checkpoint. Reapply the checkpoint reset after - // a crash in that window so stale positions cannot skip work. + // Retry and grace deferral persist the resume rewind before + // resetting the checkpoint. Repair that crash window even when + // a grace wait has never spent a failure retry. checkpoint_manager.reset_for_retry().await?; } @@ -685,6 +702,15 @@ impl ErasureSetHealer { ) -> Result<()> { // 1. get current state let state = resume_manager.get_state().await; + if self.replacement_task_id.is_some() + && let Some(retry_not_before) = state.dangling_delete_retry_not_before + && retry_not_before > SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() + { + if !state.can_retry() { + return Err(Error::ReplacementRetryBudgetExhausted); + } + return Err(Error::DanglingDeleteDeferred { retry_not_before }); + } let checkpoint = checkpoint_manager.get_checkpoint().await; debug!( @@ -956,6 +982,24 @@ impl ErasureSetHealer { if failed_objects > 0 || skipped_objects > 0 || failed_buckets > 0 { if self.replacement_task_id.is_some() { let state = resume_manager.get_state().await; + let checkpoint = checkpoint_manager.get_checkpoint().await; + if state.can_retry() + && failed_objects == 0 + && failed_buckets == 0 + && skipped_objects > 0 + && skipped_objects == checkpoint.dangling_delete_grace_objects + && !checkpoint.counter_unknown + && let Some(retry_not_before) = checkpoint.dangling_delete_retry_not_before + { + // Publish the generation's rewind and wait first. If the + // checkpoint reset is interrupted, initialization reapplies + // it without losing the deferred object classification. + resume_manager + .defer_retry_until_dangling_delete_ready(retry_not_before) + .await?; + checkpoint_manager.reset_for_retry().await?; + return Err(Error::DanglingDeleteDeferred { retry_not_before }); + } let targets_ready = self .storage .replacement_targets_ready_for_retry( @@ -1129,6 +1173,9 @@ impl ErasureSetHealer { Ok((result, Some(err))) => { let object_size = result_object_size_u64(&result); match Self::classify_heal_object_error(&err) { + HealObjectOutcome::DanglingDeleteDeferred { retry_not_before } => { + (object_size, Err(Error::DanglingDeleteDeferred { retry_not_before })) + } HealObjectOutcome::Absent | HealObjectOutcome::Transient => ( object_size, Err(Error::transient_skip(format!( @@ -1140,6 +1187,9 @@ impl ErasureSetHealer { } Err(err @ Error::TaskCancelled) | Err(err @ Error::TaskTimeout) => return Err(err), Err(err) => match Self::classify_heal_object_error(&err) { + HealObjectOutcome::DanglingDeleteDeferred { retry_not_before } => { + (0, Err(Error::DanglingDeleteDeferred { retry_not_before })) + } HealObjectOutcome::Absent | HealObjectOutcome::Transient => ( 0, Err(Error::transient_skip(format!( @@ -1170,7 +1220,8 @@ impl ErasureSetHealer { ); CheckpointObjectOutcome::Processed } - Err(Error::TransientSkip { message }) => { + Err(err) if matches!(err, Error::TransientSkip { .. } | Error::DanglingDeleteDeferred { .. }) => { + let message = err.to_string(); telemetry_unknown |= !increment_counter(counters.skipped_objects); telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size); warn!( @@ -1185,7 +1236,12 @@ impl ErasureSetHealer { error = %message, "Replacement pool metadata heal skipped due to transient error" ); - CheckpointObjectOutcome::Skipped + match err { + Error::DanglingDeleteDeferred { retry_not_before } => { + CheckpointObjectOutcome::DeferredDanglingDelete { retry_not_before } + } + _ => CheckpointObjectOutcome::Skipped, + } } Err(err) => { telemetry_unknown |= !increment_counter(counters.failed_objects); @@ -1248,6 +1304,10 @@ impl ErasureSetHealer { reason: HealDeferredReason::TransientExistenceCheck, retry_not_before: None, }, + CheckpointObjectOutcome::DeferredDanglingDelete { retry_not_before } => HealObjectDisposition::Deferred { + reason: HealDeferredReason::DanglingDeleteGrace, + retry_not_before: UNIX_EPOCH.checked_add(Duration::from_secs(retry_not_before)), + }, CheckpointObjectOutcome::Failed => HealObjectDisposition::Failed(HealFailureClass::Permanent), }; ( @@ -1669,6 +1729,9 @@ impl ErasureSetHealer { let object_size = result_object_size_u64(&result); match Self::classify_heal_object_error(&err) { HealObjectOutcome::Absent => (object_size, Ok(false)), + HealObjectOutcome::DanglingDeleteDeferred { retry_not_before } => { + (object_size, Err(Error::DanglingDeleteDeferred { retry_not_before })) + } HealObjectOutcome::Transient => ( object_size, Err(Error::transient_skip(format!( @@ -1680,6 +1743,9 @@ impl ErasureSetHealer { } Err(err) => match Self::classify_heal_object_error(&err) { HealObjectOutcome::Absent => (0, Ok(false)), + HealObjectOutcome::DanglingDeleteDeferred { retry_not_before } => { + (0, Err(Error::DanglingDeleteDeferred { retry_not_before })) + } HealObjectOutcome::Transient => ( 0, Err(Error::transient_skip(format!( @@ -1744,7 +1810,8 @@ impl ErasureSetHealer { CheckpointObjectOutcome::Processed } Err(err @ Error::TaskCancelled) | Err(err @ Error::TaskTimeout) => return Err(err), - Err(Error::TransientSkip { message }) => { + Err(err) if matches!(err, Error::TransientSkip { .. } | Error::DanglingDeleteDeferred { .. }) => { + let message = err.to_string(); telemetry_unknown |= !increment_counter(skipped_objects); telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size); demote_to_debug_when!(!take_failure_log_sample(&mut transient_skip_samples_logged), warn, target: "rustfs::heal::erasure_healer", { @@ -1759,7 +1826,12 @@ impl ErasureSetHealer { error = %message, "Erasure set object heal skipped due to transient error" }); - CheckpointObjectOutcome::Skipped + match err { + Error::DanglingDeleteDeferred { retry_not_before } => { + CheckpointObjectOutcome::DeferredDanglingDelete { retry_not_before } + } + _ => CheckpointObjectOutcome::Skipped, + } } Err(err) => { telemetry_unknown |= !increment_counter(failed_objects); @@ -2134,6 +2206,8 @@ mod resume_loop_tests { use tokio::sync::RwLock; use tokio_util::sync::CancellationToken; + mod dangling_grace; + fn item(name: &str, version: Option<&str>, delete_marker: bool) -> HealListItem { HealListItem { name: name.to_string(), @@ -2254,6 +2328,8 @@ mod resume_loop_tests { /// A transient infrastructure condition (offline disk / unmet quorum): /// the version must be recorded as skipped and retried on a later pass. Transient, + DanglingGrace(u64), + StorageError(EcstoreError), RpcCancelled(u32), Cancelled, Timeout, @@ -2475,6 +2551,10 @@ mod resume_loop_tests { Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileVersionNotFound)))) } HealOutcome::Transient => Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::DiskNotFound)))), + HealOutcome::DanglingGrace(retry_not_before) => { + Ok((HealResultItem::default(), Some(Error::DanglingDeleteDeferred { retry_not_before }))) + } + HealOutcome::StorageError(error) => Ok((HealResultItem::default(), Some(Error::Storage(error)))), HealOutcome::RpcCancelled(_) => { // Pool aggregation clones the selected error before returning it to heal. let error = EcstoreError::from(tonic::Status::cancelled("injected peer cancellation")); diff --git a/crates/heal/src/heal/erasure_healer/resume_loop_tests/dangling_grace.rs b/crates/heal/src/heal/erasure_healer/resume_loop_tests/dangling_grace.rs new file mode 100644 index 000000000..4472fdef5 --- /dev/null +++ b/crates/heal/src/heal/erasure_healer/resume_loop_tests/dangling_grace.rs @@ -0,0 +1,473 @@ +// Copyright 2026 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::*; + +#[path = "../../../../tests/storage_api.rs"] +mod real_disk_storage_api; + +use real_disk_storage_api::integration::{ObjectIO as _, WriteCompletion}; + +fn future_deadline() -> u64 { + now_secs() + 3600 +} + +fn now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("test clock must be after the epoch") + .as_secs() +} + +async fn replacement_env() -> Env { + let mut env = make_env_with_targets(vec!["replacement-a".to_owned()]).await; + env.resume + .cleanup() + .await + .expect("ordinary fixture state must be removed before binding the replacement generation"); + env.resume = ResumeManager::new_replacement_intent( + env.healer.disk.clone(), + env.task_id.clone(), + "pool_0_set_0".to_owned(), + vec!["b".to_owned()], + vec!["replacement-a".to_owned()], + vec![ReplacementTargetIdentity { + endpoint: "replacement-a".to_owned(), + canonical_path: "/mnt/replacement-a".to_owned(), + physical_device_ids: vec!["device-a".to_owned()], + filesystem_identity: "filesystem-a".to_owned(), + }], + ) + .await + .expect("replacement intent should persist before scanning"); + env.healer = ErasureSetHealer::new( + env.storage.clone(), + Arc::new(RwLock::new(HealProgress::new())), + CancellationToken::new(), + env.healer.disk.clone(), + HealOpts::default(), + HealRequestSource::AutoHeal, + ) + .with_replacement_targets(vec!["replacement-a".to_owned()], Some(env.task_id.clone())); + env.storage + .set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME)); + env +} + +async fn execute(env: &Env) -> Result<()> { + env.healer + .execute_heal_with_resume(&["b".to_owned()], "pool_0_set_0", &env.resume, &env.checkpoint) + .await +} + +fn grace_page(env: &Env, deadline: u64) { + env.storage.set_page( + None, + Page { + items: vec![item("dangling", Some("v1"), false)], + next: None, + truncated: false, + }, + ); + env.storage + .set_outcome("dangling", Some("v1"), HealOutcome::DanglingGrace(deadline)); + env.storage + .set_result("dangling", Some("v1"), replacement_target_ok_result("replacement-a", "dangling")); +} + +async fn persist_partial_grace_page(env: &Env, deadline: u64) { + env.checkpoint + .record_object_outcome(CheckpointObjectOutcomeRecord { + object: compose_key("dangling", Some("v1")), + outcome: CheckpointObjectOutcome::DeferredDanglingDelete { + retry_not_before: deadline, + }, + successful: 0, + failed: 0, + skipped: 1, + bytes: 0, + skipped_new_versions: 0, + skipped_ilm_expired: 0, + counter_unknown: false, + }) + .await + .expect("deferred object outcome should be recorded atomically"); + env.checkpoint + .advance_page(0, 1) + .await + .expect("checkpoint should survive a crash before resume cursor publication"); +} + +#[tokio::test] +async fn dangling_grace_pass_preserves_replacement_retry_budget() { + let env = replacement_env().await; + env.storage.set_page( + None, + Page { + items: vec![item("dangling", Some("v1"), false)], + next: None, + truncated: false, + }, + ); + env.storage + .set_outcome("dangling", Some("v1"), HealOutcome::DanglingGrace(future_deadline())); + + execute(&env) + .await + .expect_err("dangling cleanup must remain pending during the grace window"); + + let state = ResumeManager::load_replacement_intent(env.healer.disk.clone(), &env.task_id) + .await + .expect("grace-protected replacement should remain recoverable") + .get_state() + .await; + assert_eq!( + state.retry_count, 0, + "waiting for the grace window must not spend the replacement retry budget" + ); + assert!(!state.completed, "a deferred object must prevent replacement completion"); + assert_eq!(env.storage.calls(), vec![("dangling".to_owned(), Some("v1".to_owned()))]); +} + +#[tokio::test] +async fn dangling_grace_restart_waits_without_replaying_and_finishes_after_expiration() { + let mut env = replacement_env().await; + let deadline = future_deadline(); + grace_page(&env, deadline); + let error = execute(&env).await.expect_err("the first pass must defer cleanup"); + assert!(matches!(error, Error::DanglingDeleteDeferred { retry_not_before } if retry_not_before == deadline)); + + for _ in 0..4 { + env.resume = ResumeManager::load_replacement_intent(env.healer.disk.clone(), &env.task_id) + .await + .expect("each restarted executor should reload the same replacement generation"); + env.checkpoint = CheckpointManager::load_from_disk(env.healer.disk.clone(), &env.task_id) + .await + .expect("each restarted executor should reload the durable checkpoint"); + let error = execute(&env).await.expect_err("early execution must preserve the grace wait"); + assert!(matches!(error, Error::DanglingDeleteDeferred { retry_not_before } if retry_not_before == deadline)); + let state = env.resume.get_state().await; + assert_eq!(state.retry_count, 0); + assert_eq!(state.dangling_delete_retry_not_before, Some(deadline)); + assert!(!state.completed); + } + assert_eq!( + env.storage.calls(), + vec![("dangling".to_owned(), Some("v1".to_owned()))], + "restart and early attempts must not replay protected cleanup" + ); + + env.resume + .defer_retry_until_dangling_delete_ready(now_secs().saturating_sub(1)) + .await + .expect("an expired durable wait should retain the same failure budget"); + env.storage.set_outcome("dangling", Some("v1"), HealOutcome::Ok); + env.healer + .heal_erasure_set(&["b".to_owned()], "pool_0_set_0") + .await + .expect("expiration must allow a fresh scan to verify the replacement"); + let state = ResumeManager::load_replacement_intent(env.healer.disk.clone(), &env.task_id) + .await + .expect("verified replacement intent must survive until marker cleanup") + .get_state() + .await; + assert!(state.completed); + assert_eq!(state.replacement_phase, crate::heal::resume::ReplacementPhase::Verified); + assert_eq!(state.retry_count, 0); + assert_eq!(state.dangling_delete_retry_not_before, None); + assert_eq!( + env.storage.calls().iter().filter(|(object, _)| object == "dangling").count(), + 2, + "completion must follow a new object heal after expiration" + ); +} + +#[tokio::test] +async fn dangling_grace_partial_page_checkpoint_restores_reason_without_rehealing_identity() { + let mut env = replacement_env().await; + let deadline = future_deadline(); + grace_page(&env, deadline); + env.storage.set_page( + None, + Page { + items: vec![item("dangling", Some("v1"), false), item("healthy", Some("v1"), false)], + next: None, + truncated: false, + }, + ); + env.storage + .set_result("healthy", Some("v1"), replacement_target_ok_result("replacement-a", "healthy")); + persist_partial_grace_page(&env, deadline).await; + env.resume = ResumeManager::load_replacement_intent(env.healer.disk.clone(), &env.task_id) + .await + .expect("resume state should survive the partial-page crash"); + env.checkpoint = CheckpointManager::load_from_disk(env.healer.disk.clone(), &env.task_id) + .await + .expect("partial-page grace reason must be durable"); + let checkpoint = env.checkpoint.get_checkpoint().await; + assert_eq!(checkpoint.dangling_delete_grace_objects, 1); + assert_eq!(checkpoint.dangling_delete_retry_not_before, Some(deadline)); + + let error = execute(&env) + .await + .expect_err("a checkpointed grace outcome must still defer the whole pass"); + assert!(matches!(error, Error::DanglingDeleteDeferred { retry_not_before } if retry_not_before == deadline)); + let state = env.resume.get_state().await; + assert_eq!(state.retry_count, 0, "checkpoint replay must retain the budget exemption"); + assert_eq!(state.dangling_delete_retry_not_before, Some(deadline)); + assert_eq!(env.storage.calls(), vec![("healthy".to_owned(), Some("v1".to_owned()))]); +} + +#[tokio::test] +async fn dangling_grace_resume_reset_recovers_the_stale_checkpoint_before_rescan() { + let env = replacement_env().await; + let deadline = now_secs().saturating_sub(1); + grace_page(&env, deadline); + persist_partial_grace_page(&env, deadline).await; + env.checkpoint + .update_position(4, 9) + .await + .expect("a stale scan position should persist before the resume reset"); + env.resume + .defer_retry_until_dangling_delete_ready(deadline) + .await + .expect("publish the resume reset before simulating a checkpoint reset crash"); + + let (resumed, checkpoint) = env + .healer + .initialize_resume_state(&env.task_id, "pool_0_set_0", &["b".to_owned()]) + .await + .expect("initialization should reconcile both persistence layers after the crash"); + let snapshot = checkpoint.get_checkpoint().await; + assert_eq!(snapshot.current_bucket_index, 0); + assert_eq!(snapshot.current_object_index, 0); + assert!( + snapshot.skipped_objects.is_empty(), + "stale dedup cannot suppress the deferred version after expiration" + ); + assert_eq!(snapshot.dangling_delete_grace_objects, 0); + assert_eq!(snapshot.dangling_delete_retry_not_before, None); + assert_eq!(resumed.get_state().await.retry_count, 0); + + env.storage.set_outcome("dangling", Some("v1"), HealOutcome::Ok); + env.healer + .execute_heal_with_resume(&["b".to_owned()], "pool_0_set_0", &resumed, &checkpoint) + .await + .expect("repaired checkpoint must let the expired version heal again"); + assert!(env.storage.calls().contains(&("dangling".to_owned(), Some("v1".to_owned())))); +} + +#[tokio::test] +async fn dangling_grace_mixed_transient_failures_still_exhaust_the_bounded_budget() { + let env = replacement_env().await; + grace_page(&env, future_deadline()); + env.storage.set_page( + None, + Page { + items: vec![item("dangling", Some("v1"), false), item("offline", Some("v1"), false)], + next: None, + truncated: false, + }, + ); + env.storage.set_outcome("offline", Some("v1"), HealOutcome::Transient); + let max_retries = env.resume.get_state().await.max_retries; + for expected in 1..=max_retries { + let error = execute(&env) + .await + .expect_err("ordinary infrastructure failures must keep the bounded retry path"); + assert!(matches!(error, Error::TransientSkip { .. })); + let state = env.resume.get_state().await; + assert_eq!(state.retry_count, expected); + assert_eq!( + state.dangling_delete_retry_not_before, None, + "grace cannot hide the ordinary failure budget" + ); + assert!(!state.completed); + } + execute(&env) + .await + .expect_err("a grace object must not reopen an exhausted ordinary failure budget"); + let state = env.resume.get_state().await; + assert_eq!(state.retry_count, max_retries); + assert_eq!(state.dangling_delete_retry_not_before, None); + assert!(!state.completed); +} + +#[tokio::test] +async fn dangling_grace_expiration_still_rejects_a_remounted_replacement_target() { + let mut env = replacement_env().await; + grace_page(&env, future_deadline()); + execute(&env).await.expect_err("initial grace should defer the replacement"); + env.resume + .defer_retry_until_dangling_delete_ready(now_secs().saturating_sub(1)) + .await + .expect("make the same generation ready for a new scan"); + env.storage.set_outcome("dangling", Some("v1"), HealOutcome::Ok); + let expected = env.resume.get_state().await.replacement_target_identities; + let mut remounted = expected.clone(); + remounted[0].physical_device_ids = vec!["device-b".to_owned()]; + remounted[0].filesystem_identity = "filesystem-b".to_owned(); + env.storage + .replacement_target_identity_sequences + .lock() + .expect("identity fixture") + .push_back(remounted); + env.healer = env.healer.with_replacement_identity_fence(Some(expected)); + let calls_before = env.storage.calls(); + let listings_before = env.storage.list_include_lifecycle_object_info_calls(); + + let error = execute(&env) + .await + .expect_err("expiration cannot admit another physical replacement identity"); + assert!( + matches!(&error, Error::TransientSkip { message } if message.contains("1 bucket(s) failed")), + "the set-level result must retain the failed identity-fenced bucket: {error}" + ); + assert!( + env.storage + .replacement_target_identity_sequences + .lock() + .expect("identity fixture") + .is_empty(), + "the new scan must check the remounted target identity" + ); + assert_eq!( + env.storage.list_include_lifecycle_object_info_calls(), + listings_before, + "the identity fence must reject the remount before listing its object page" + ); + assert_eq!( + env.storage.calls(), + calls_before, + "a remount must be rejected before replaying any object" + ); + let state = env.resume.get_state().await; + assert!(!state.completed); + assert_ne!(state.replacement_phase, crate::heal::resume::ReplacementPhase::Verified); +} + +#[test] +#[serial_test::serial] +fn dangling_grace_real_storage_error_persists_a_rounded_deadline_without_spending_budget() { + // Match the real-storage integration fixtures' stack budget for composed heal futures. + const STACK_SIZE: usize = 8 * 1024 * 1024; + std::thread::Builder::new() + .name("dangling-grace-raw-storage".to_owned()) + .stack_size(STACK_SIZE) + .spawn(|| { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .thread_stack_size(STACK_SIZE) + .enable_all() + .build() + .expect("real-storage grace runtime should build"); + runtime.block_on(temp_env::async_with_vars( + [("RUSTFS_HEAL_DANGLING_DELETE_GRACE_SECS", Some("3600"))], + real_storage_grace_round_trip(), + )); + }) + .expect("real-storage grace test thread should spawn") + .join() + .expect("real-storage grace test thread should finish"); +} + +async fn real_storage_grace_round_trip() { + use crate::heal::storage::{ECStoreHealStorage, HealObjectOptions, HealPutObjReader}; + + let directory = TempDir::new().expect("real-storage grace fixture directory"); + let real = rustfs_test_utils::TestECStoreEnv::builder() + .base_dir(directory.path()) + .disk_count(4) + .build() + .await; + let bucket = "typed-grace-fixture"; + let object = "dangling"; + real.make_bucket(bucket, false).await; + real.ecstore + .put_object( + bucket, + object, + &mut HealPutObjReader::from_vec(b"recent inline shard".to_vec()), + &HealObjectOptions { + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + .expect("real PUT must finish its rename tails before simulating missing shards"); + let metadata_path = real.disk_paths[0].join(bucket).join(object).join("xl.meta"); + let metadata_before = std::fs::read(&metadata_path).expect("capture the surviving recent inline metadata"); + for disk in &real.disk_paths[1..] { + std::fs::remove_dir_all(disk.join(bucket).join(object)).expect("remove three completed copies below the read quorum"); + } + let storage = ECStoreHealStorage::new(real.ecstore.clone()); + let (_, error) = storage + .heal_object( + bucket, + object, + None, + &HealOpts { + scan_mode: rustfs_heal_contracts::heal_channel::HealScanMode::Deep, + recreate: true, + pool: Some(0), + set: Some(0), + ..Default::default() + }, + ) + .await + .expect("real dangling cleanup should return a typed heal result"); + let Error::Storage(error) = error.expect("the surviving recent shard must be grace-protected") else { + panic!("real ECStore heal must retain its Storage error wrapper"); + }; + assert!(error.is_dangling_delete_grace(), "the fixture must return a real grace marker: {error}"); + let retry_after = error + .dangling_delete_retry_after() + .expect("the real grace error must retain typed retry timing") + .as_secs(); + assert!(retry_after > 0, "the new object must have a remaining grace window"); + assert_eq!( + std::fs::read(&metadata_path).expect("read protected metadata after real heal"), + metadata_before + ); + + let env = replacement_env().await; + grace_page(&env, future_deadline()); + env.storage + .set_outcome("dangling", Some("v1"), HealOutcome::StorageError(error)); + let before = now_secs(); + let result = execute(&env).await.expect_err("raw storage grace must defer the replacement"); + let after = now_secs(); + let Error::DanglingDeleteDeferred { retry_not_before } = result else { + panic!("raw storage grace must become a typed absolute-deadline wait: {result}"); + }; + assert!( + (before + retry_after + 1..=after + retry_after + 1).contains(&retry_not_before), + "classification must round the raw whole-second delay up at the epoch boundary" + ); + let state = ResumeManager::load_replacement_intent(env.healer.disk.clone(), &env.task_id) + .await + .expect("reload the raw-error grace wait from the survivor anchor") + .get_state() + .await; + assert_eq!( + state.retry_count, 0, + "raw storage errors must use the same budget exemption as typed waits" + ); + assert_eq!(state.dangling_delete_retry_not_before, Some(retry_not_before)); + assert!(!state.completed); + assert_eq!( + std::fs::read(&metadata_path).expect("read surviving metadata after classification"), + metadata_before + ); +} diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index b087e9f1d..b998b7bda 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -594,10 +594,36 @@ fn completed_status_is_retrying(status: &HealTaskStatus) -> bool { matches!(status, HealTaskStatus::Retrying { .. }) } +fn automatic_replacement_grace_deadline(task: &HealTask, result: &Result<()>) -> Option { + if task.source != HealRequestSource::AutoHeal + || !matches!(task.heal_type, HealType::ErasureSet { .. }) + || task.heal_endpoints.is_empty() + { + return None; + } + match result { + Err(Error::DanglingDeleteDeferred { retry_not_before }) => Some(*retry_not_before), + _ => None, + } +} + fn retry_budget_for_result(task: &HealTask, result: &Result<()>, retryable_batch_failure: bool) -> Option<(Duration, String)> { let Err(err) = result else { return None; }; + if let Some(retry_not_before) = automatic_replacement_grace_deadline(task, result) { + // This typed result is emitted only while the durable generation has + // retry budget. A storage grace period is waiting, not a failed attempt. + let now = SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + let delay = Duration::from_secs(retry_not_before.saturating_sub(now)); + // Persisted deadlines are untrusted. Reject a duration that Tokio's + // monotonic timer cannot represent instead of panicking in sleep. + tokio::time::Instant::now().checked_add(delay)?; + return Some((delay, err.to_string())); + } if task.retry_attempts >= MAX_RECOVERABLE_HEAL_RETRIES { return None; } @@ -621,13 +647,17 @@ fn retry_budget_for_result(task: &HealTask, result: &Result<()>, retryable_batch #[cfg(test)] fn retry_request_for_result(task: &HealTask, result: &Result<()>) -> Option<(HealRequest, Duration, String)> { let (delay, error) = retry_budget_for_result(task, result, false)?; - Some((task.retry_request(), delay, error)) + let mut request = task.retry_request(); + if automatic_replacement_grace_deadline(task, result).is_some() { + request.retry_attempts = task.retry_attempts; + } + Some((request, delay, error)) } async fn retry_request_for_result_with_budget(task: &HealTask, result: &Result<()>) -> Option<(HealRequest, Duration, String)> { let retryable_batch_failure = task.batch_failure_is_retryable().await; let (delay, error) = retry_budget_for_result(task, result, retryable_batch_failure)?; - let request = match task.retry_request_with_remaining_timeout().await { + let mut request = match task.retry_request_with_remaining_timeout().await { Ok(request) => request, Err(err) => { debug!( @@ -643,6 +673,9 @@ async fn retry_request_for_result_with_budget(task: &HealTask, result: &Result<( return None; } }; + if automatic_replacement_grace_deadline(task, result).is_some() { + request.retry_attempts = task.retry_attempts; + } Some((request, delay, error)) } diff --git a/crates/heal/src/heal/manager/auto_scan.rs b/crates/heal/src/heal/manager/auto_scan.rs index ef59857ce..b09f337db 100644 --- a/crates/heal/src/heal/manager/auto_scan.rs +++ b/crates/heal/src/heal/manager/auto_scan.rs @@ -459,6 +459,13 @@ impl HealManager { } }; let mut state = resume_manager.get_state().await; + let now = SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs(); + if durable_replacement_recovery_is_waiting_for_grace(&state, &task_id, now) { + // Waiting retains generation ownership without + // blocking its set as a recovery conflict. + deferred_recovery_sets.insert(state.set_disk_id.clone()); + continue; + } if replacement_retry_is_exhausted_active(&state) { match resume_manager .rearm_replacement_recovery_if_needed(storage.as_ref()) diff --git a/crates/heal/src/heal/manager/scheduler.rs b/crates/heal/src/heal/manager/scheduler.rs index 7641a772d..2a02fc6b8 100644 --- a/crates/heal/src/heal/manager/scheduler.rs +++ b/crates/heal/src/heal/manager/scheduler.rs @@ -14,6 +14,21 @@ /// The heal scheduler: queue consumption loop and its skip/metric helpers. use super::*; +pub(super) fn retry_admission_delay( + grace_deadline: Option, + ordinary_delay: Duration, + grace_admission_backoff: Duration, + now: u64, +) -> Option { + let delay = match grace_deadline { + Some(deadline) if deadline > now => Duration::from_secs(deadline - now), + Some(_) => grace_admission_backoff, + None => ordinary_delay, + }; + tokio::time::Instant::now().checked_add(delay)?; + Some(delay) +} + impl HealManager { /// Start scheduler pub(super) async fn start_scheduler(&self) -> Result<()> { @@ -458,6 +473,7 @@ impl HealManager { if let (Some((retry_request, retry_delay, retry_error)), Some(retry_cancel_token)) = (retry_request_for_queue, retry_cancel_token) { + let grace_deadline = automatic_replacement_grace_deadline(task.as_ref(), &result); let retry_request_id = retry_request.id.clone(); let retry_attempt = retry_request.retry_attempts; let retry_key = PriorityHealQueue::make_dedup_key(&retry_request); @@ -473,7 +489,18 @@ impl HealManager { let retry_manager_cancel_token = manager_cancel_token.clone(); let retry_config = config_for_spawn.clone(); tokio::spawn(async move { + let mut grace_admission_backoff = Duration::ZERO; loop { + let now = SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + let Some(admission_delay) = + retry_admission_delay(grace_deadline, retry_delay, grace_admission_backoff, now) + else { + retrying_heals_for_spawn.lock().await.remove(&retry_request_id); + return; + }; tokio::select! { _ = retry_cancel_token.cancelled() => { debug!( @@ -493,7 +520,17 @@ impl HealManager { retrying_heals_for_spawn.lock().await.remove(&retry_request_id); return; } - _ = sleep(retry_delay) => {} + _ = sleep(admission_delay) => {} + } + + // A wall-clock rollback during the monotonic + // wait must not admit work before its deadline. + let now = SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + if grace_deadline.is_some_and(|deadline| deadline > now) { + continue; } { @@ -582,12 +619,12 @@ impl HealManager { event = EVENT_HEAL_QUEUE_ADMISSION, component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_MANAGER, + result = "retry_enqueued", request_id = %retry_request_id, priority = ?retry_priority, retry_attempt, - retry_delay_ms = retry_delay.as_millis(), + retry_delay_ms = admission_delay.as_millis(), error = %retry_error, - result = "retry_enqueued", "Heal retry admission decided" ); if should_notify { @@ -631,10 +668,11 @@ impl HealManager { event = EVENT_HEAL_QUEUE_ADMISSION, component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_MANAGER, + result = "retry_rejected_full", request_id = %retry_request_id, priority = ?retry_priority, retry_attempt, - result = "retry_rejected_full", + retry_delay_ms = admission_delay.as_millis(), "Heal retry admission decided" ); } @@ -644,15 +682,20 @@ impl HealManager { event = EVENT_HEAL_QUEUE_ADMISSION, component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_MANAGER, + result = "retry_dropped", request_id = %retry_request_id, priority = ?retry_priority, retry_attempt, + retry_delay_ms = admission_delay.as_millis(), reason = reason.as_str(), - result = "retry_dropped", "Heal retry admission decided" ); } } + // Once the storage deadline has passed, queue + // pressure uses ordinary bounded admission + // backoff instead of repeating the grace wait. + grace_admission_backoff = recoverable_heal_retry_delay(retry_attempt); } }); } diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 251ef3f0f..997c85e64 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -2750,6 +2750,177 @@ async fn contended_healing_marker_cas_retains_bounded_task_retries() { ); } +#[tokio::test] +async fn automatic_replacement_grace_waits_preserve_request_retry_budget() { + let now = SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("test clock should follow the epoch") + .as_secs(); + let retry_not_before = now + 3600; + let result = Err(Error::DanglingDeleteDeferred { retry_not_before }); + let mut request = HealRequest::new( + HealType::ErasureSet { + buckets: vec!["bucket".to_string()], + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + request.source = HealRequestSource::AutoHeal; + request.heal_endpoints = vec!["replacement-a".to_string()]; + request.retry_attempts = 1; + let original = request.clone(); + let storage: Arc = Arc::new(MockStorage); + + for _ in 0..MAX_RECOVERABLE_HEAL_RETRIES * 3 { + let task = HealTask::from_request(request, storage.clone()); + let (retry, delay, _) = retry_request_for_result_with_budget(&task, &result) + .await + .expect("legal grace waiting must survive more passes than the failure retry limit"); + assert_eq!(retry.retry_attempts, original.retry_attempts); + assert_eq!(retry.id, original.id); + assert_eq!(retry.heal_type, original.heal_type); + assert_eq!(retry.source, original.source); + assert_eq!(retry.heal_endpoints, original.heal_endpoints); + assert!( + delay > MAX_RECOVERABLE_HEAL_RETRY_DELAY, + "grace waiting must respect the storage deadline" + ); + assert!(delay <= Duration::from_secs(3600)); + request = retry; + } + + let ordinary_failure = Err(Error::transient_skip("replacement quorum is unavailable")); + for attempt in original.retry_attempts + 1..=MAX_RECOVERABLE_HEAL_RETRIES { + let task = HealTask::from_request(request, storage.clone()); + let (retry, _, _) = retry_request_for_result_with_budget(&task, &ordinary_failure) + .await + .expect("a genuine failure should still consume the remaining retry budget"); + assert_eq!(retry.retry_attempts, attempt); + request = retry; + } + let task = HealTask::from_request(request, storage); + assert!(retry_request_for_result_with_budget(&task, &ordinary_failure).await.is_none()); +} + +#[test] +fn automatic_replacement_grace_uses_absolute_deadline_and_keeps_task_scope() { + let mut request = HealRequest::new( + HealType::ErasureSet { + buckets: vec!["bucket".to_string()], + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions::default(), + HealPriority::Low, + ); + request.source = HealRequestSource::AutoHeal; + request.heal_endpoints = vec!["replacement-a".to_string()]; + request.retry_attempts = MAX_RECOVERABLE_HEAL_RETRIES; + let expired = Err(Error::DanglingDeleteDeferred { retry_not_before: 1 }); + let task = HealTask::from_request(request.clone(), Arc::new(MockStorage)); + let (retry, delay, _) = retry_request_for_result(&task, &expired) + .expect("an elapsed storage deadline can requeue the typed in-budget generation immediately"); + assert_eq!(delay, Duration::ZERO); + assert_eq!(retry.retry_attempts, request.retry_attempts); + + let unrepresentable = Err(Error::DanglingDeleteDeferred { + retry_not_before: u64::MAX, + }); + assert!( + retry_request_for_result(&task, &unrepresentable).is_none(), + "an unrepresentable persisted deadline must never reach the scheduler timer" + ); + + request.heal_endpoints.clear(); + let task = HealTask::from_request(request.clone(), Arc::new(MockStorage)); + assert!( + retry_request_for_result(&task, &expired).is_none(), + "ordinary set healing must retain its retry cap" + ); + request.heal_endpoints = vec!["replacement-a".to_string()]; + request.source = HealRequestSource::Admin; + let task = HealTask::from_request(request, Arc::new(MockStorage)); + assert!( + retry_request_for_result(&task, &expired).is_none(), + "admin healing must retain its retry cap" + ); +} + +#[tokio::test] +async fn automatic_replacement_grace_wait_excludes_execution_timeout_budget() { + let now = SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("test clock should follow the epoch") + .as_secs(); + let result = Err(Error::DanglingDeleteDeferred { + retry_not_before: now + 3600, + }); + let mut request = HealRequest::new( + HealType::ErasureSet { + buckets: vec!["bucket".to_string()], + set_disk_id: "pool_0_set_0".to_string(), + }, + HealOptions { + timeout: Some(Duration::from_secs(rustfs_config::DEFAULT_HEAL_TASK_TIMEOUT_SECS)), + ..HealOptions::default() + }, + HealPriority::Low, + ); + request.source = HealRequestSource::AutoHeal; + request.heal_endpoints = vec!["replacement-a".to_string()]; + let task = HealTask::from_request(request, Arc::new(MockStorage)); + task.set_execution_elapsed_for_test(Duration::from_secs(30)).await; + + let (retry, delay, _) = retry_request_for_result_with_budget(&task, &result) + .await + .expect("a grace wait longer than the execution timeout must remain schedulable"); + let remaining = retry.options.timeout.expect("execution timeout must remain configured"); + assert!(remaining > Duration::from_secs(269)); + assert!(remaining <= Duration::from_secs(270)); + assert!(delay > remaining, "the storage wait must not be deducted from the execution budget"); + + let waiting_task = HealTask::from_request(retry, Arc::new(MockStorage)); + let (next_retry, _, _) = retry_request_for_result_with_budget(&waiting_task, &result) + .await + .expect("an unstarted retry must retain its unused execution timeout"); + assert_eq!(next_retry.options.timeout, Some(remaining)); + assert_eq!(next_retry.retry_attempts, 0); +} + +#[test] +fn automatic_replacement_grace_admission_rechecks_deadline_after_queue_pressure() { + let original_grace_delay = Duration::from_secs(3600); + let queue_backoff = recoverable_heal_retry_delay(0); + assert_eq!( + retry_admission_delay(Some(3700), original_grace_delay, Duration::ZERO, 100), + Some(original_grace_delay), + "the first admission must wait until the storage deadline" + ); + assert_eq!( + retry_admission_delay(Some(3700), original_grace_delay, Duration::ZERO, 3700), + Some(Duration::ZERO), + "the first admission may proceed as soon as the deadline expires" + ); + for now in [3700, 3800, 7200] { + assert_eq!( + retry_admission_delay(Some(3700), original_grace_delay, queue_backoff, now), + Some(queue_backoff), + "a full or dropped queue admission must not repeat the hour-long grace wait" + ); + } + assert_eq!( + retry_admission_delay(Some(3700), original_grace_delay, queue_backoff, 3690), + Some(Duration::from_secs(10)), + "a clock rollback must restore the wait for the absolute storage deadline" + ); + assert_eq!( + retry_admission_delay(None, Duration::from_secs(8), queue_backoff, 7200), + Some(Duration::from_secs(8)), + "ordinary failure retries must retain their original admission backoff" + ); + assert!(retry_admission_delay(Some(u64::MAX), original_grace_delay, queue_backoff, 100).is_none()); +} + #[tokio::test] async fn retry_request_for_result_preserves_remaining_timeout_budget() { let storage: Arc = Arc::new(MockStorage); @@ -2893,6 +3064,60 @@ fn durable_replacement_recovery_re_admits_only_the_matching_generation() { ); } +#[test] +fn durable_replacement_grace_waiting_preserves_reservation_until_deadline() { + let task_id = "grace-generation"; + let mut state = ResumeState::new( + task_id.to_string(), + "erasure_set".to_string(), + "pool_0_set_0".to_string(), + vec!["bucket".to_string()], + ); + state.replacement_generation = Some(task_id.to_string()); + state.replacement_phase = ReplacementPhase::Rebuilding; + state.replacement_targets = vec!["replacement-a".to_string()]; + state.dangling_delete_retry_not_before = Some(200); + + for now in [100, 150, 199] { + assert!( + durable_replacement_recovery_is_waiting_for_grace(&state, task_id, now), + "scan and startup replay must defer this original generation before the deadline" + ); + assert!( + durable_replacement_reserves_targets(&state), + "waiting must prevent competing replacement admission" + ); + assert!( + durable_replacement_recovery_is_due(&state, task_id), + "legal waiting must remain eligible rather than blocked" + ); + } + for now in [200, 201] { + assert!(!durable_replacement_recovery_is_waiting_for_grace(&state, task_id, now)); + assert!( + durable_replacement_recovery_is_due(&state, task_id), + "the same generation must resume when its deadline expires" + ); + } + + state.retry_count = state.max_retries; + assert!(durable_replacement_reserves_targets(&state)); + assert!( + !durable_replacement_recovery_is_waiting_for_grace(&state, task_id, 100), + "a stale deadline must not hide a genuinely exhausted generation" + ); + assert!(!durable_replacement_recovery_is_due(&state, task_id)); + + state.completed = true; + state.replacement_phase = ReplacementPhase::CleanupPending; + assert!(durable_replacement_recovery_is_due(&state, task_id)); + assert!( + !durable_replacement_recovery_is_waiting_for_grace(&state, task_id, 100), + "verified marker cleanup must not be postponed by a stale storage deadline" + ); + assert!(!durable_replacement_recovery_is_waiting_for_grace(&state, "other-generation", 100)); +} + #[test] fn replacement_target_reservation_ends_only_after_an_explicit_transfer() { let mut state = ResumeState::new( diff --git a/crates/heal/src/heal/manager/unclean_shutdown.rs b/crates/heal/src/heal/manager/unclean_shutdown.rs index 9dae1c499..45396568a 100644 --- a/crates/heal/src/heal/manager/unclean_shutdown.rs +++ b/crates/heal/src/heal/manager/unclean_shutdown.rs @@ -46,6 +46,12 @@ pub(super) fn durable_replacement_recovery_is_due(state: &ResumeState, task_id: && matches!(state.replacement_phase, ReplacementPhase::Verified | ReplacementPhase::CleanupPending))) } +pub(super) fn durable_replacement_recovery_is_waiting_for_grace(state: &ResumeState, task_id: &str, now: u64) -> bool { + !state.completed + && durable_replacement_recovery_is_due(state, task_id) + && state.dangling_delete_retry_not_before.is_some_and(|deadline| deadline > now) +} + pub(super) fn replacement_discovery_error_is_expected_for_deferred_endpoint( error: &Error, endpoint: &str, @@ -371,6 +377,15 @@ impl HealManager { } }; let mut state = manager.get_state().await; + let now = SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + if durable_replacement_recovery_is_waiting_for_grace(&state, &task_id, now) { + // Discovery already reserved this set. Keep the original + // generation and its markers until the storage deadline. + continue; + } if replacement_retry_is_exhausted_active(&state) { match manager.rearm_replacement_recovery_if_needed(self.storage.as_ref()).await { Ok(true) => state = manager.get_state().await, diff --git a/crates/heal/src/heal/resume.rs b/crates/heal/src/heal/resume.rs index 7c2e2ab4a..b97617b2d 100644 --- a/crates/heal/src/heal/resume.rs +++ b/crates/heal/src/heal/resume.rs @@ -76,6 +76,14 @@ const REPLACEMENT_RECOVERY_CORRUPTION_PREFIX: &str = "replacement recovery corru /// latest. Discard their cursor and progress and scan from the beginning. const CURRENT_RESUME_SCHEMA: u32 = 7; +fn dangling_delete_deadline_is_representable(deadline: u64) -> bool { + let Some(not_before) = UNIX_EPOCH.checked_add(Duration::from_secs(deadline)) else { + return false; + }; + let remaining = not_before.duration_since(SystemTime::now()).unwrap_or_default(); + Instant::now().checked_add(remaining).is_some() +} + /// Persistence throttle for per-object bookkeeping: flush after this many /// buffered mutations or once the interval elapses, whichever comes first. /// Object heal is idempotent, so a crash re-heals at most one throttle window. @@ -361,6 +369,10 @@ pub struct ResumeState { /// back into a consumed retry attempt. #[serde(default)] pub replacement_retry_waiting_for_target: bool, + /// A pure dangling-delete grace pass waits until this UNIX deadline + /// without consuming the generation's failure retry budget. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub dangling_delete_retry_not_before: Option, /// start time pub start_time: u64, /// last update time @@ -439,6 +451,7 @@ impl ResumeState { replacement_legacy_successor: None, replacement_legacy_retry_compatibility_done: false, replacement_retry_waiting_for_target: false, + dangling_delete_retry_not_before: None, start_time: SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(), last_update: SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(), completed: false, @@ -557,6 +570,7 @@ impl ResumeState { pub fn mark_completed(&mut self) { self.completed = true; + self.dangling_delete_retry_not_before = None; self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(); } @@ -868,6 +882,14 @@ impl ResumeManager { message: "Resume state task id does not match filename".to_string(), }); } + if state + .dangling_delete_retry_not_before + .is_some_and(|deadline| !dangling_delete_deadline_is_representable(deadline)) + { + return Err(Error::TaskExecutionFailed { + message: "Resume state has an unrepresentable dangling-delete grace deadline".to_string(), + }); + } // Older progress cannot prove that the exact null slot was inspected. if state.schema_version > CURRENT_RESUME_SCHEMA { @@ -916,6 +938,7 @@ impl ResumeManager { state.baseline_generation = None; state.baseline_known = false; state.counter_unknown = false; + state.dangling_delete_retry_not_before = None; state.completed = false; state.pending_buckets.append(&mut state.completed_buckets); state.pending_buckets.sort(); @@ -1149,6 +1172,7 @@ impl ResumeManager { return Ok(false); } state.replacement_retry_waiting_for_target = false; + state.dangling_delete_retry_not_before = None; state.increment_retry(); state.reset_for_retry(); drop(state); @@ -1156,6 +1180,26 @@ impl ResumeManager { Ok(true) } + /// Rewind a pure grace-wait pass without spending a failure retry. The + /// durable deadline also fences checkpoint reset recovery after a crash. + pub(crate) async fn defer_retry_until_dangling_delete_ready(&self, retry_not_before: u64) -> Result<()> { + if !dangling_delete_deadline_is_representable(retry_not_before) { + return Err(Error::TaskExecutionFailed { + message: "Cannot defer to an unrepresentable dangling-delete grace deadline".to_string(), + }); + } + let mut state = self.state.write().await; + if !state.can_retry() { + return Err(Error::ReplacementRetryBudgetExhausted); + } + state.replacement_retry_waiting_for_target = false; + state.dangling_delete_retry_not_before = Some(retry_not_before); + state.reset_for_retry(); + state.error_message = Some(Error::DanglingDeleteDeferred { retry_not_before }.to_string()); + drop(state); + self.save_state_strict().await + } + /// Rewind one replacement pass while its target is not ready, without /// consuming the generation retry budget. The caller resets the companion /// checkpoint first; this method rewinds the resume ledger and keeps the @@ -1177,6 +1221,7 @@ impl ResumeManager { state.replacement_legacy_retry_compatibility_done = true; } state.replacement_retry_waiting_for_target = true; + state.dangling_delete_retry_not_before = None; state.reset_for_retry(); state.error_message = Some(REPLACEMENT_TARGET_READINESS_DEFERRED.to_string()); drop(state); diff --git a/crates/heal/src/heal/resume/checkpoint.rs b/crates/heal/src/heal/resume/checkpoint.rs index 9ca4de2ad..50d27150c 100644 --- a/crates/heal/src/heal/resume/checkpoint.rs +++ b/crates/heal/src/heal/resume/checkpoint.rs @@ -26,7 +26,7 @@ use super::super::storage_api::owner::{EcstoreConditionalFileUpdate, EcstoreDisk use super::super::{BUCKET_META_PREFIX, DiskStore, HealDiskExt, RUSTFS_META_BUCKET}; use super::{ LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, PersistThrottle, RESUME_CHECKPOINT_BLOCKED_FILE, RESUME_CHECKPOINT_FILE, - delete_resume_file, path_to_str, validate_resume_task_id, + dangling_delete_deadline_is_representable, delete_resume_file, path_to_str, validate_resume_task_id, }; const EVENT_HEAL_CHECKPOINT_STATE: &str = "heal_checkpoint_state"; @@ -44,6 +44,11 @@ pub enum CheckpointObjectOutcome { Processed, Failed, Skipped, + DeferredDanglingDelete { retry_not_before: u64 }, +} + +fn u64_is_zero(value: &u64) -> bool { + *value == 0 } #[derive(Debug)] @@ -127,6 +132,11 @@ pub struct ResumeCheckpoint { pub failed_object_count: u64, #[serde(default)] pub skipped_object_count: u64, + // Omit defaults to preserve the integrity digest of older checkpoints. + #[serde(default, skip_serializing_if = "u64_is_zero")] + pub dangling_delete_grace_objects: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub dangling_delete_retry_not_before: Option, #[serde(default)] pub skipped_new_versions: u64, #[serde(default)] @@ -166,6 +176,8 @@ impl ResumeCheckpoint { successful_objects: 0, failed_object_count: 0, skipped_object_count: 0, + dangling_delete_grace_objects: 0, + dangling_delete_retry_not_before: None, skipped_new_versions: 0, skipped_ilm_expired: 0, processed_bytes: 0, @@ -243,6 +255,8 @@ impl ResumeCheckpoint { self.successful_objects = 0; self.failed_object_count = 0; self.skipped_object_count = 0; + self.dangling_delete_grace_objects = 0; + self.dangling_delete_retry_not_before = None; self.skipped_new_versions = 0; self.skipped_ilm_expired = 0; self.processed_bytes = 0; @@ -407,6 +421,13 @@ impl CheckpointManager { message: "Resume checkpoint task id does not match filename".to_string(), }); } + if checkpoint + .dangling_delete_retry_not_before + .is_some_and(|deadline| !dangling_delete_deadline_is_representable(deadline)) + { + Self::block_invalid_snapshot(&disk, task_id).await; + return Err(Error::InvalidCheckpoint("Unrepresentable dangling-delete grace deadline".to_string())); + } // Older checkpoints can contain identities that are not comparable to // the current keys or lack their corresponding aggregate counters. @@ -484,6 +505,8 @@ impl CheckpointManager { checkpoint.successful_objects = 0; checkpoint.failed_object_count = 0; checkpoint.skipped_object_count = 0; + checkpoint.dangling_delete_grace_objects = 0; + checkpoint.dangling_delete_retry_not_before = None; checkpoint.skipped_new_versions = 0; checkpoint.skipped_ilm_expired = 0; checkpoint.processed_bytes = 0; @@ -658,6 +681,22 @@ impl CheckpointManager { CheckpointObjectOutcome::Processed => checkpoint.add_processed_object(object), CheckpointObjectOutcome::Failed => checkpoint.add_failed_object(object), CheckpointObjectOutcome::Skipped => checkpoint.add_skipped_object(object), + CheckpointObjectOutcome::DeferredDanglingDelete { retry_not_before } => { + if !dangling_delete_deadline_is_representable(retry_not_before) { + return Err(Error::InvalidCheckpoint("Unrepresentable dangling-delete grace deadline".to_string())); + } + if checkpoint.skipped_objects.insert(object) { + checkpoint.dangling_delete_grace_objects = checkpoint + .dangling_delete_grace_objects + .checked_add(1) + .ok_or_else(|| Error::InvalidCheckpoint("Dangling-delete grace counter overflow".to_string()))?; + checkpoint.dangling_delete_retry_not_before = Some( + checkpoint + .dangling_delete_retry_not_before + .map_or(retry_not_before, |deadline| deadline.max(retry_not_before)), + ); + } + } } checkpoint.update_progress(successful, failed, skipped, bytes); checkpoint.set_skipped_version_counts(skipped_new_versions, skipped_ilm_expired); diff --git a/crates/heal/src/heal/resume/replacement.rs b/crates/heal/src/heal/resume/replacement.rs index f8f99f107..2c920a739 100644 --- a/crates/heal/src/heal/resume/replacement.rs +++ b/crates/heal/src/heal/resume/replacement.rs @@ -506,6 +506,22 @@ impl ResumeManager { if state.replacement_generation.as_deref() != Some(state.task_id.as_str()) { return Err(replacement_recovery_conflict("replacement failure has no matching generation")); } + if let Error::DanglingDeleteDeferred { retry_not_before } = error { + if !super::dangling_delete_deadline_is_representable(*retry_not_before) { + return Err(Error::TaskExecutionFailed { + message: "Cannot persist an unrepresentable dangling-delete grace deadline".to_string(), + }); + } + if !state.can_retry() { + return Err(Error::ReplacementRetryBudgetExhausted); + } + state.dangling_delete_retry_not_before = Some(*retry_not_before); + state.replacement_retry_waiting_for_target = false; + state.error_message = Some(error.to_string()); + state.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(); + drop(state); + return self.save_state_strict().await; + } let readiness_deferred = matches!(error, Error::ReplacementTargetNotReady(_)) || (state.replacement_retry_waiting_for_target && matches!(error, Error::TransientSkip { message } if message.contains("target readiness deferred retry"))); @@ -514,6 +530,7 @@ impl ResumeManager { state.retry_count = state.max_retries.saturating_sub(1); } state.replacement_retry_waiting_for_target = true; + state.dangling_delete_retry_not_before = None; state.error_message = Some(super::REPLACEMENT_TARGET_READINESS_DEFERRED.to_string()); state.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(); drop(state); diff --git a/crates/heal/src/heal/resume/tests.rs b/crates/heal/src/heal/resume/tests.rs index 0cfe022b9..bc4cdc303 100644 --- a/crates/heal/src/heal/resume/tests.rs +++ b/crates/heal/src/heal/resume/tests.rs @@ -116,6 +116,7 @@ fn replacement_recovery_merges_same_generation_replicas() { let mut latest = replacement_candidate_state(&task_id); latest.replacement_revision = 5; latest.processed_objects = 7; + latest.dangling_delete_retry_not_before = Some(4_000_000_000); merge_replacement_recovery_candidate( &mut current, ReplacementRecoveryCandidate::new(latest, "http://survivor-b/recovery", true).expect("valid candidate"), @@ -125,6 +126,7 @@ fn replacement_recovery_merges_same_generation_replicas() { let selected = current.expect("a canonical candidate should remain"); assert_eq!(selected.state.replacement_revision, 5); assert_eq!(selected.state.processed_objects, 7); + assert_eq!(selected.state.dangling_delete_retry_not_before, Some(4_000_000_000)); assert_eq!(selected.anchor, "http://survivor-b/recovery"); } @@ -149,6 +151,21 @@ fn replacement_recovery_rejects_ambiguous_equal_revision_copies() { assert!(matches!(error, Error::ReplacementGenerationConflict { .. })); } +#[test] +fn replacement_recovery_rejects_divergent_grace_deadlines_at_the_same_revision() { + let task_id = Uuid::new_v4().to_string(); + let left = replacement_candidate_state(&task_id); + let mut right = left.clone(); + right.dangling_delete_retry_not_before = Some(4_000_000_000); + let mut current = Some(ReplacementRecoveryCandidate::new(left, "survivor-a", true).expect("valid candidate")); + let error = merge_replacement_recovery_candidate( + &mut current, + ReplacementRecoveryCandidate::new(right, "survivor-b", false).expect("valid deferred candidate"), + ) + .expect_err("different deadlines at the same revision must remain a conflict"); + assert!(matches!(error, Error::ReplacementGenerationConflict { .. })); +} + #[test] fn replacement_recovery_rejects_different_generation_binding() { let task_id = Uuid::new_v4().to_string(); @@ -253,6 +270,7 @@ async fn replacement_terminal_phases_are_durable() { ) .await .expect("replacement intent should persist"); + manager.state.write().await.dangling_delete_retry_not_before = Some(4_000_000_000); manager .mark_replacement_completed_and_verified() .await @@ -265,6 +283,7 @@ async fn replacement_terminal_phases_are_durable() { .await; assert!(verified.completed); assert_eq!(verified.replacement_phase, ReplacementPhase::Verified); + assert_eq!(verified.dangling_delete_retry_not_before, None); let resumed = ResumeManager::new_replacement_intent( disk.clone(), @@ -1547,6 +1566,275 @@ fn reset_for_retry_clears_progress_but_keeps_retry_budget() { assert_eq!(state.retry_count, 1, "retry budget must be preserved"); } +#[tokio::test] +async fn dangling_delete_deferral_survives_restart_without_spending_retry_budget() { + let (_temp_dir, disk) = schema_test_disk().await; + let task_id = ResumeUtils::generate_task_id(); + let manager = ResumeManager::new( + disk.clone(), + task_id.clone(), + "erasure_set".to_string(), + "pool_0_set_0".to_string(), + vec!["bucket".to_string()], + ) + .await + .expect("create resume state for dangling-delete deferral"); + { + let mut state = manager.state.write().await; + state.processed_objects = 9; + state.skipped_objects = 4; + state.retry_count = 2; + state.resume_cursor = Some("unfinished-page".to_string()); + } + let deadline = 4_000_000_000; + manager + .defer_retry_until_dangling_delete_ready(deadline) + .await + .expect("persist dangling-delete deferral"); + + let resumed = ResumeManager::load_from_disk(disk, &task_id) + .await + .expect("reload dangling-delete deferral after restart"); + let state = resumed.get_state().await; + assert_eq!(state.retry_count, 2, "grace waits must preserve the failure retry budget"); + assert_eq!(state.dangling_delete_retry_not_before, Some(deadline)); + assert_eq!(state.processed_objects, 0); + assert_eq!(state.skipped_objects, 0); + assert!(state.resume_cursor.is_none()); + assert!(!state.completed); + + assert!(resumed.schedule_retry().await.expect("schedule an ordinary failure retry")); + let state = resumed.get_state().await; + assert_eq!(state.retry_count, 3); + assert_eq!(state.dangling_delete_retry_not_before, None); +} + +#[tokio::test] +async fn replacement_dangling_delete_failure_preserves_budget_and_does_not_rearm_exhaustion() { + let (_temp_dir, disk) = schema_test_disk().await; + let task_id = ResumeUtils::generate_task_id(); + let initial = replacement_candidate_state(&task_id); + let manager = ResumeManager::new_replacement_intent( + disk.clone(), + task_id.clone(), + initial.set_disk_id, + initial.replacement_buckets, + initial.replacement_targets, + initial.replacement_target_identities, + ) + .await + .expect("persist replacement intent for grace failure"); + { + let mut state = manager.state.write().await; + state.replacement_phase = ReplacementPhase::Rebuilding; + state.retry_count = 1; + } + manager.save_state_strict().await.expect("persist initial retry count"); + let deferred = Error::DanglingDeleteDeferred { + retry_not_before: 4_000_000_000, + }; + manager + .record_replacement_failure(&deferred, 99) + .await + .expect("persist grace failure without a task-level retry increment"); + let reloaded = ResumeManager::load_replacement_intent(disk.clone(), &task_id) + .await + .expect("reload replacement grace failure"); + let state = reloaded.get_state().await; + assert_eq!(state.retry_count, 1); + assert_eq!(state.dangling_delete_retry_not_before, Some(4_000_000_000)); + assert_eq!(state.replacement_phase, ReplacementPhase::Rebuilding); + { + let mut state = reloaded.state.write().await; + state.retry_count = state.max_retries; + state.dangling_delete_retry_not_before = None; + } + reloaded.save_state_strict().await.expect("persist exhausted retry budget"); + assert!(matches!( + reloaded.record_replacement_failure(&deferred, 99).await, + Err(Error::ReplacementRetryBudgetExhausted) + )); + let exhausted = ResumeManager::load_replacement_intent(disk, &task_id) + .await + .expect("reload unchanged exhausted intent") + .get_state() + .await; + assert_eq!(exhausted.retry_count, exhausted.max_retries); + assert_eq!(exhausted.dangling_delete_retry_not_before, None); + assert!(matches!( + reloaded.defer_retry_until_dangling_delete_ready(4_000_000_000).await, + Err(Error::ReplacementRetryBudgetExhausted) + )); +} + +#[tokio::test] +async fn old_signed_checkpoint_accepts_missing_dangling_delete_fields() { + use sha2::{Digest, Sha256}; + + let (_temp_dir, disk) = schema_test_disk().await; + let task_id = ResumeUtils::generate_task_id(); + let mut checkpoint = ResumeCheckpoint::new(task_id.clone()); + checkpoint.current_object_index = 1; + checkpoint.successful_objects = 1; + checkpoint.processed_objects.insert("bucket/object:v1".to_string()); + let mut wire = serde_json::to_value(&checkpoint).expect("serialize old checkpoint fixture"); + let fields = wire.as_object_mut().expect("checkpoint should be an object"); + fields.remove("dangling_delete_grace_objects"); + fields.remove("dangling_delete_retry_not_before"); + let unsigned = serde_json::to_vec(&wire).expect("serialize old unsigned checkpoint fixture"); + wire["integrity_digest"] = serde_json::json!(base64_simd::STANDARD.encode_to_string(Sha256::digest(&unsigned))); + let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}"); + disk.write_all( + RUSTFS_META_BUCKET, + &checkpoint_path, + serde_json::to_vec(&wire) + .expect("serialize old signed checkpoint fixture") + .into(), + ) + .await + .expect("persist old signed checkpoint fixture"); + + let restored = CheckpointManager::load_from_disk(disk.clone(), &task_id) + .await + .expect("old signed checkpoints must preserve their original digest") + .get_checkpoint() + .await; + assert_eq!(restored.current_object_index, 1); + assert_eq!(restored.successful_objects, 1); + assert_eq!(restored.dangling_delete_grace_objects, 0); + assert_eq!(restored.dangling_delete_retry_not_before, None); + assert!(!CheckpointManager::is_blocked(&disk, &task_id).await); +} + +#[tokio::test] +async fn checkpoint_grace_classification_survives_page_commit_and_restart() { + let (_temp_dir, disk) = schema_test_disk().await; + let task_id = ResumeUtils::generate_task_id(); + let checkpoint = CheckpointManager::new(disk.clone(), task_id.clone()) + .await + .expect("create checkpoint for grace outcomes"); + for (object, deadline, skipped) in [ + ("bucket/a:v1", 4_000_000_000, 1), + ("bucket/b:v1", 4_000_000_100, 2), + ("bucket/b:v1", 4_000_000_200, 2), + ] { + checkpoint + .record_object_outcome(CheckpointObjectOutcomeRecord { + object: object.to_string(), + outcome: CheckpointObjectOutcome::DeferredDanglingDelete { + retry_not_before: deadline, + }, + successful: 0, + failed: 0, + skipped, + bytes: 0, + skipped_new_versions: 0, + skipped_ilm_expired: 0, + counter_unknown: false, + }) + .await + .expect("record grace outcome atomically with object counters"); + } + checkpoint + .advance_page(0, 2) + .await + .expect("flush page containing grace outcomes"); + checkpoint + .prune_completed_page() + .await + .expect("prune page identities after cursor commit"); + + let restored = CheckpointManager::load_from_disk(disk.clone(), &task_id) + .await + .expect("reload grace page after restart"); + let snapshot = restored.get_checkpoint().await; + assert_eq!(snapshot.skipped_object_count, 2); + assert_eq!( + snapshot.dangling_delete_grace_objects, 2, + "duplicate page outcomes must not increase grace count" + ); + assert_eq!(snapshot.dangling_delete_retry_not_before, Some(4_000_000_100)); + assert!(snapshot.skipped_objects.is_empty(), "classification must outlive page identity pruning"); + + restored.reset_for_retry().await.expect("reset checkpoint for grace recheck"); + let reset = CheckpointManager::load_from_disk(disk, &task_id) + .await + .expect("reload checkpoint after reset") + .get_checkpoint() + .await; + assert_eq!(reset.dangling_delete_grace_objects, 0); + assert_eq!(reset.dangling_delete_retry_not_before, None); + assert_eq!(reset.skipped_object_count, 0); + assert_eq!(reset.current_object_index, 0); +} + +#[test] +fn old_schema_seven_resume_defaults_dangling_delete_deadline() { + let task_id = ResumeUtils::generate_task_id(); + let mut wire = serde_json::to_value(replacement_candidate_state(&task_id)).expect("serialize replacement intent"); + wire.as_object_mut() + .expect("replacement intent should be an object") + .remove("dangling_delete_retry_not_before"); + let restored: ResumeState = serde_json::from_value(wire).expect("decode old schema-seven intent"); + assert_eq!(restored.schema_version, 7); + assert_eq!(restored.dangling_delete_retry_not_before, None); + assert!(restored.replacement_legacy_retry_compatibility_done); +} + +#[tokio::test] +async fn persisted_dangling_delete_deadlines_reject_unrepresentable_values() { + use sha2::{Digest, Sha256}; + + let (_temp_dir, disk) = schema_test_disk().await; + let task_id = ResumeUtils::generate_task_id(); + let mut state = ResumeState::new( + task_id.clone(), + "erasure_set".to_string(), + "pool_0_set_0".to_string(), + vec!["bucket".to_string()], + ); + state.dangling_delete_retry_not_before = Some(u64::MAX); + let state_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_STATE_FILE}"); + disk.write_all( + RUSTFS_META_BUCKET, + &state_path, + serde_json::to_vec(&state) + .expect("serialize malformed deadline fixture") + .into(), + ) + .await + .expect("write resume state with an unrepresentable deadline"); + let resume_error = match ResumeManager::load_from_disk(disk.clone(), &task_id).await { + Ok(_) => panic!("an unrepresentable deadline must not become a durable wait"), + Err(error) => error, + }; + assert!(matches!(resume_error, Error::TaskExecutionFailed { message } if message.contains("grace deadline"))); + + let mut checkpoint = ResumeCheckpoint::new(task_id.clone()); + checkpoint.dangling_delete_grace_objects = 1; + checkpoint.skipped_object_count = 1; + checkpoint.dangling_delete_retry_not_before = Some(u64::MAX); + let mut wire = serde_json::to_value(checkpoint).expect("serialize malformed checkpoint deadline"); + let unsigned = serde_json::to_vec(&wire).expect("serialize unsigned malformed checkpoint"); + wire["integrity_digest"] = serde_json::json!(base64_simd::STANDARD.encode_to_string(Sha256::digest(&unsigned))); + let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}"); + disk.write_all( + RUSTFS_META_BUCKET, + &checkpoint_path, + serde_json::to_vec(&wire) + .expect("serialize signed malformed checkpoint") + .into(), + ) + .await + .expect("write signed checkpoint with an unrepresentable deadline"); + let checkpoint_error = match CheckpointManager::load_from_disk(disk.clone(), &task_id).await { + Ok(_) => panic!("an unrepresentable checkpoint deadline must not become a durable wait"), + Err(error) => error, + }; + assert!(matches!(checkpoint_error, Error::InvalidCheckpoint(message) if message.contains("grace deadline"))); + assert!(CheckpointManager::is_blocked(&disk, &task_id).await); +} + #[tokio::test] async fn replacement_readiness_deferral_rewinds_progress_without_spending_budget() { let (_temp_dir, disk) = schema_test_disk().await; @@ -1572,6 +1860,7 @@ async fn replacement_readiness_deferral_rewinds_progress_without_spending_budget state.processed_objects = 9; state.skipped_objects = 4; state.retry_count = 2; + state.dangling_delete_retry_not_before = Some(4_000_000_000); } manager .save_state_strict() @@ -1593,6 +1882,7 @@ async fn replacement_readiness_deferral_rewinds_progress_without_spending_budget assert_eq!(state.replacement_phase, ReplacementPhase::Rebuilding); assert!(!state.completed); assert!(state.replacement_retry_waiting_for_target); + assert_eq!(state.dangling_delete_retry_not_before, None); assert_eq!(state.error_message.as_deref(), Some(REPLACEMENT_TARGET_READINESS_DEFERRED)); }