diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index d92af72f9..53bfa729a 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -671,7 +671,9 @@ impl ErasureSetHealer { error = %e, "Erasure set bucket heal failed" ); - // continue to next bucket, do not interrupt the whole process + // A single durable cursor and ledger cannot safely preserve + // this bucket while processing a later one. + break; } } @@ -1233,7 +1235,13 @@ impl ErasureSetHealer { if !is_truncated { break; } - resume_manager.set_resume_cursor(next_token.clone()).await?; + continuation_token = next_heal_listing_token(bucket, "", next_token, is_truncated)?; + if continuation_token.is_none() { + // A truncated page without a continuation token is terminal. + // Retain its ledger until bucket completion is durable. + break; + } + resume_manager.set_resume_cursor(continuation_token.clone()).await?; checkpoint_manager.prune_completed_page().await?; // Anti-loop guard: an empty page reported as truncated cannot advance @@ -1253,12 +1261,6 @@ impl ErasureSetHealer { ))); } previous_page_last = page_last; - - continuation_token = next_heal_listing_token(bucket, "", next_token, is_truncated)?; - if continuation_token.is_none() { - // Truncated but no continuation token: treat as end of listing. - break; - } } Ok(()) @@ -1645,6 +1647,7 @@ mod resume_loop_tests { list_include_lifecycle_object_info: Mutex>, replacement_target_identity_sequences: Mutex>>, fail_listing: AtomicBool, + fail_listing_buckets: Mutex>, } impl FakeStorage { @@ -1681,6 +1684,9 @@ mod resume_loop_tests { fn fail_listing(&self) { self.fail_listing.store(true, Ordering::SeqCst); } + fn fail_bucket_listing(&self, bucket: &str) { + self.fail_listing_buckets.lock().unwrap().insert(bucket.to_string()); + } } #[async_trait::async_trait] @@ -1771,7 +1777,7 @@ mod resume_loop_tests { } async fn list_objects_for_heal_page( &self, - _bucket: &str, + bucket: &str, _prefix: &str, continuation_token: Option<&str>, include_lifecycle_object_info: bool, @@ -1780,7 +1786,7 @@ mod resume_loop_tests { .lock() .unwrap() .push(include_lifecycle_object_info); - if self.fail_listing.load(Ordering::SeqCst) { + if self.fail_listing.load(Ordering::SeqCst) || self.fail_listing_buckets.lock().unwrap().contains(bucket) { return Err(Error::other("injected listing failure")); } let key = continuation_token.map(str::to_string); @@ -2158,6 +2164,49 @@ mod resume_loop_tests { assert!(state.completed_buckets.is_empty(), "the failed bucket must remain resumable"); } + #[tokio::test] + async fn bucket_failure_stops_before_a_later_bucket_checkpoint() { + let env = make_env().await; + let task_id = ResumeUtils::generate_task_id(); + let buckets = vec!["a".to_string(), "b".to_string()]; + let resume = ResumeManager::new( + env.healer.disk.clone(), + task_id.clone(), + "erasure_set".to_string(), + "pool_0_set_0".to_string(), + buckets.clone(), + ) + .await + .unwrap(); + let checkpoint = CheckpointManager::new(env.healer.disk.clone(), task_id.clone()) + .await + .unwrap(); + env.storage.fail_bucket_listing("a"); + for _ in 0..3 { + assert!(resume.schedule_retry().await.unwrap()); + } + + env.healer + .execute_heal_with_resume(&buckets, "pool_0_set_0", &resume, &checkpoint) + .await + .expect_err("the first bucket failure must keep the pass incomplete"); + let persisted = checkpoint.get_checkpoint().await; + assert_eq!(persisted.current_bucket_index, 0); + assert!(resume.get_state().await.completed_buckets.is_empty()); + + let resumed = ResumeManager::load_from_disk(env.healer.disk.clone(), &task_id) + .await + .unwrap(); + let checkpoint = CheckpointManager::load_from_disk(env.healer.disk.clone(), &task_id) + .await + .unwrap(); + env.healer + .execute_heal_with_resume(&buckets, "pool_0_set_0", &resumed, &checkpoint) + .await + .expect_err("recovery must retry the earlier failed bucket"); + assert!(!resumed.get_state().await.completed); + } + #[tokio::test] async fn completed_resume_state_is_not_selected_for_a_new_heal() { let env = make_env().await; @@ -2452,6 +2501,39 @@ mod resume_loop_tests { assert_eq!(state.processed_objects, 2); } + #[tokio::test] + async fn truncated_page_without_token_retains_its_replay_ledger() { + let env = make_env().await; + env.storage.set_page( + None, + Page { + items: vec![item("object", Some("v1"), false)], + next: None, + truncated: true, + }, + ); + + let (processed, successful, failed, skipped, result) = run(&env).await; + result.expect("the tokenless truncated page is a terminal page"); + assert_eq!((processed, successful, failed, skipped), (1, 1, 0, 0)); + + let resumed = ResumeManager::load_from_disk(env.healer.disk.clone(), &env.task_id) + .await + .unwrap(); + let checkpoint = CheckpointManager::load_from_disk(env.healer.disk.clone(), &env.task_id) + .await + .unwrap(); + env.healer + .execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &resumed, &checkpoint) + .await + .expect("terminal-page recovery must not replay a durable identity"); + + assert_eq!(env.storage.calls(), vec![("object".to_string(), Some("v1".to_string()))]); + let state = resumed.get_state().await; + assert_eq!(state.successful_objects, 1); + assert_eq!(state.processed_objects, 1); + } + #[tokio::test] async fn completed_bucket_reconciles_its_final_page_checkpoint_after_crash() { let env = make_env().await;