fix(heal): defer replacement retries until dangling grace expires (#8388)

This commit is contained in:
cxymds
2026-10-07 22:04:12 +08:00
committed by GitHub
parent 19978b2cb6
commit 6b1554003e
12 changed files with 1292 additions and 18 deletions
+8 -1
View File
@@ -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<uuid::Uuid> },
@@ -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<std::time::SystemTime> {
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),
+89 -9
View File
@@ -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"));
@@ -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
);
}
+35 -2
View File
@@ -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<u64> {
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))
}
@@ -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())
+48 -5
View File
@@ -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<u64>,
ordinary_delay: Duration,
grace_admission_backoff: Duration,
now: u64,
) -> Option<Duration> {
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);
}
});
}
+225
View File
@@ -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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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(
@@ -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,
+45
View File
@@ -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<u64>,
/// 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);
+40 -1
View File
@@ -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<u64>,
#[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);
@@ -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);
+290
View File
@@ -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));
}