gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): repair all object versions during disk-replacement heal (#4400)

Browse Source 
Disk-replacement heal previously repaired only the latest version of each
object and never enumerated objects whose latest version is a delete marker,
so old versions were left unrepaired on a replaced drive.

Switch heal enumeration from list_objects_v2 (latest-only) to
list_object_versions (every version incl. delete markers), thread the concrete
version_id into the existing per-version heal_object, and make resume
cursor-based instead of positional: an opaque (marker, version_marker) paging
token persisted in ResumeState, a length-prefixed injective per-version dedup
key, schema_version bumps (v2) migrated independently in each of the two
persisted files, and a retry that resets both managers together (fixing a
latent rescan-skips-everything defect). Adds a real-disk-wipe e2e regression
suite proving old versions and delete-marker-latest objects are physically
restored.

Fixes rustfs/backlog#918
Fixes rustfs/backlog#919
Closes rustfs/backlog#854
This commit is contained in:
Zhengchao An
2026-07-08 08:58:52 +08:00
committed by GitHub
parent 4715dbed01
commit 2b063b0c4a
11 changed files with 1831 additions and 488 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -9261,6 +9261,7 @@ name = "rustfs-heal"
version = "1.0.0-beta.8"
dependencies = [
"async-trait",
"base64 0.22.1",
"futures",
"http 1.4.2",
"metrics",
crates/heal/Cargo.toml
+1
View File
@@ -44,6 +44,7 @@ uuid = { workspace = true, features = ["v4", "serde"] }
async-trait = { workspace = true }
futures = { workspace = true }
metrics = { workspace = true }
base64 = { workspace = true }
[dev-dependencies]
serde_json = { workspace = true }
crates/heal/src/heal/channel.rs
+6 -2
View File
@@ -606,7 +606,11 @@ mod tests {
) -> crate::Result<(rustfs_madmin::heal_commands::HealResultItem, Option<crate::Error>)> {
Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None))
}
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> crate::Result<Vec<String>> {
async fn list_objects_for_heal(
&self,
_bucket: &str,
_prefix: &str,
) -> crate::Result<Vec<crate::heal::storage::HealListItem>> {
Ok(vec![])
}
async fn list_objects_for_heal_page(
@@ -614,7 +618,7 @@ mod tests {
_bucket: &str,
_prefix: &str,
_continuation_token: Option<&str>,
) -> crate::Result<(Vec<String>, Option<String>, bool)> {
) -> crate::Result<(Vec<crate::heal::storage::HealListItem>, Option<String>, bool)> {
Ok((vec![], None, false))
}
async fn get_disk_for_resume(&self, _set_disk_id: &str) -> crate::Result<DiskStore> {
crates/heal/src/heal/erasure_healer.rs
+517 -328
View File
@@ -14,12 +14,12 @@
use crate::heal::{
progress::HealProgress,
resume::{CheckpointManager, ResumeManager, ResumeUtils},
resume::{CheckpointManager, ResumeManager, ResumeUtils, compose_key},
storage::{HealStorageAPI, next_heal_listing_token},
task::is_missing_object_dir_heal_result,
};
use crate::{Error, Result};
use futures::{StreamExt, future::join_all, stream::FuturesUnordered};
use futures::{StreamExt, stream::FuturesUnordered};
use metrics::gauge;
use rustfs_common::heal_channel::{HealOpts, HealRequestSource, HealScanMode};
use std::sync::{
@@ -27,7 +27,7 @@ use std::sync::{
atomic::{AtomicUsize, Ordering},
};
use tokio::sync::{RwLock, Semaphore};
use tracing::{debug, error, info, warn};
use tracing::{debug, error, warn};
use super::{DiskStore, EcstoreError};
@@ -447,6 +447,12 @@ impl ErasureSetHealer {
// (backlog#855 / #799 B6).
if failed_objects > 0 {
if resume_manager.schedule_retry().await? {
// Both persistence layers must be reset together: schedule_retry
// rewinds the resume state (cursor + counters), and the
// checkpoint's per-version dedup sets + position must be cleared
// in lockstep, or the retry would skip the very versions it is
// meant to re-heal.
checkpoint_manager.reset_for_retry().await?;
// Retry budget remains: state has been reset for a full re-scan.
// Return Err so `heal_erasure_set` preserves (does not clean up)
// the resume/checkpoint state and the caller keeps the healing
@@ -549,15 +555,23 @@ impl ErasureSetHealer {
}
};
// 2. process objects with pagination to avoid loading all objects into memory
let mut continuation_token: Option<String> = None;
// 2. process object VERSIONS with pagination to avoid loading everything into memory.
// The continuation token is the authoritative opaque (marker, version_marker)
// cursor seeded from the resume state, so a resume continues exactly where the
// previous pass stopped — including mid-object across version pages.
let mut continuation_token: Option<String> = resume_manager.resume_cursor().await;
let mut global_obj_idx = 0usize;
// Anti-loop guard: the last (name, version_id) actually observed on the
// previous page. Comparing decoded item identities (not raw tokens)
// detects a non-advancing cursor even though next_marker embeds a
// transient cache id that changes between identical pages.
let mut previous_page_last: Option<(String, Option<String>)> = None;
let page_concurrency_limit =
Self::effective_heal_page_object_concurrency_for_source(self.source, self.heal_opts.scan_mode);
let in_flight = Arc::new(AtomicUsize::new(0));
loop {
// Get one page of objects
// Get one page of object versions
let (objects, next_token, is_truncated) = self
.storage
.list_objects_for_heal_page(bucket, "", continuation_token.as_deref())
@@ -567,30 +581,33 @@ impl ErasureSetHealer {
let semaphore = Arc::new(Semaphore::new(page_concurrency_limit));
let mut page_tasks = FuturesUnordered::new();
for object in objects {
let object_idx = global_obj_idx;
// Capture the last version identity of this page for the anti-loop guard.
let page_last = objects.last().map(|item| (item.name.clone(), item.version_id.clone()));
for item in objects {
// current_object_index is now only a progress metric; the cursor
// drives resume, so we no longer skip by position.
global_obj_idx += 1;
if object_idx < *current_object_index {
continue;
}
if checkpoint.processed_objects.contains(&object) || checkpoint.skipped_objects.contains(&object) {
// Per-version dedup identity — the single canonical key.
let key = compose_key(&item.name, item.version_id.as_deref());
if checkpoint.processed_objects.contains(&key) || checkpoint.skipped_objects.contains(&key) {
continue;
}
resume_manager
.set_current_item(Some(bucket.to_string()), Some(object.clone()))
.set_current_item(Some(bucket.to_string()), Some(item.name.clone()))
.await?;
let storage = self.storage.clone();
let bucket_name = bucket.to_string();
let object_name = object.clone();
let object_name = item.name.clone();
let version_id = item.version_id.clone();
let dedup_key = key;
let cancel_token = self.cancel_token.clone();
let in_flight = in_flight.clone();
let set_label = set_disk_id.to_string();
let heal_opts = self.heal_opts;
let deep_scan = matches!(heal_opts.scan_mode, HealScanMode::Deep);
let semaphore = semaphore.clone();
page_tasks.push(async move {
@@ -602,7 +619,7 @@ impl ErasureSetHealer {
let _permit = match permit {
Ok(permit) => permit,
Err(err) => return (object_name, Err(err)),
Err(err) => return (dedup_key, object_name, version_id, Err(err)),
};
let current_in_flight = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
@@ -612,11 +629,20 @@ impl ErasureSetHealer {
)
.set(current_in_flight as f64);
// Always go through heal_object. Genuine absence flows through
// heal_object -> FileVersionNotFound/FileNotFound ->
// classify_heal_object_error -> Absent, so gone versions are
// recorded as skipped-ok rather than failed. The delete-marker
// vs data path is chosen internally in ops/heal.rs.
let result = if cancel_token.is_cancelled() {
Err(Error::TaskCancelled)
} else if deep_scan {
match storage.heal_object(&bucket_name, &object_name, None, &heal_opts).await {
} else {
match storage
.heal_object(&bucket_name, &object_name, version_id.as_deref(), &heal_opts)
.await
{
Ok((_result, None)) => Ok(true),
Ok((_, Some(err))) if is_missing_object_dir_heal_result(&object_name, &err) => Ok(false),
Ok((_, Some(err))) | Err(err) => match Self::classify_heal_object_error(&err) {
HealObjectOutcome::Absent => Ok(false),
HealObjectOutcome::Transient => Err(Error::transient_skip(format!(
@@ -625,51 +651,6 @@ impl ErasureSetHealer {
HealObjectOutcome::Failed => Err(err),
},
}
} else {
let object_exists = match storage.object_exists(&bucket_name, &object_name).await {
Ok(exists) => exists,
Err(err @ Error::TransientSkip { .. }) => {
let current = in_flight.fetch_sub(1, Ordering::SeqCst) - 1;
gauge!(
"rustfs_heal_page_concurrency_current",
"set" => set_label.clone()
)
.set(current as f64);
return (object_name, Err(err));
}
Err(err) => {
let object_name_for_error = object_name.clone();
let current = in_flight.fetch_sub(1, Ordering::SeqCst) - 1;
gauge!(
"rustfs_heal_page_concurrency_current",
"set" => set_label.clone()
)
.set(current as f64);
return (
object_name,
Err(Error::other(format!(
"Failed to check existence of {}/{}: {}",
bucket_name, object_name_for_error, err
))),
);
}
};
if !object_exists {
Ok(false)
} else {
match storage.heal_object(&bucket_name, &object_name, None, &heal_opts).await {
Ok((_result, None)) => Ok(true),
Ok((_, Some(err))) if is_missing_object_dir_heal_result(&object_name, &err) => Ok(false),
Ok((_, Some(err))) | Err(err) => match Self::classify_heal_object_error(&err) {
HealObjectOutcome::Absent => Ok(false),
HealObjectOutcome::Transient => Err(Error::transient_skip(format!(
"Skipped heal for {bucket_name}/{object_name} due to transient error: {err}"
))),
HealObjectOutcome::Failed => Err(err),
},
}
}
};
let current = in_flight.fetch_sub(1, Ordering::SeqCst) - 1;
@@ -679,16 +660,16 @@ impl ErasureSetHealer {
)
.set(current as f64);
(object_name, result)
(dedup_key, object_name, version_id, result)
});
}
let mut completed_in_page = 0usize;
while let Some((object, result)) = page_tasks.next().await {
while let Some((key, object, version_id, result)) = page_tasks.next().await {
match result {
Ok(true) => {
*successful_objects += 1;
checkpoint_manager.add_processed_object(object.clone()).await?;
checkpoint_manager.add_processed_object(key).await?;
debug!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
@@ -697,12 +678,13 @@ impl ErasureSetHealer {
set_disk_id,
bucket,
object = %object,
version_id = ?version_id,
state = "healed",
"Erasure set object healed"
);
}
Ok(false) => {
checkpoint_manager.add_processed_object(object.clone()).await?;
checkpoint_manager.add_processed_object(key).await?;
*successful_objects += 1;
debug!(
target: "rustfs::heal::erasure_healer",
@@ -712,6 +694,7 @@ impl ErasureSetHealer {
set_disk_id,
bucket,
object = %object,
version_id = ?version_id,
state = "missing_treated_as_ok",
"Erasure set missing object treated as ok"
);
@@ -726,7 +709,7 @@ impl ErasureSetHealer {
}
Err(Error::TransientSkip { message }) => {
*skipped_objects += 1;
checkpoint_manager.add_skipped_object(object.clone()).await?;
checkpoint_manager.add_skipped_object(key).await?;
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
@@ -735,6 +718,7 @@ impl ErasureSetHealer {
set_disk_id,
bucket,
object = %object,
version_id = ?version_id,
state = "transient_skip",
error = %message,
"Erasure set object heal skipped due to transient error"
@@ -742,7 +726,7 @@ impl ErasureSetHealer {
}
Err(err) => {
*failed_objects += 1;
checkpoint_manager.add_failed_object(object.clone()).await?;
checkpoint_manager.add_failed_object(key).await?;
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
@@ -751,6 +735,7 @@ impl ErasureSetHealer {
set_disk_id,
bucket,
object = %object,
version_id = ?version_id,
state = "failed",
error = %err,
"Erasure set object heal failed"
@@ -767,6 +752,12 @@ impl ErasureSetHealer {
}
*current_object_index = global_obj_idx;
// Persist the authoritative cursor FIRST (points at the next page
// boundary), then prune the per-version dedup sets. Both are
// idempotent under crash: heal_object re-heals safely.
let next_cursor = if is_truncated { next_token.clone() } else { None };
resume_manager.set_resume_cursor(next_cursor.clone()).await?;
checkpoint_manager.complete_page(bucket_index, *current_object_index).await?;
gauge!(
"rustfs_heal_page_concurrency_current",
@@ -779,7 +770,20 @@ impl ErasureSetHealer {
break;
}
// Anti-loop guard: if the backend keeps reporting truncation but the
// last version identity did not advance, we would spin forever.
if page_last.is_some() && page_last == previous_page_last {
return Err(Error::other(format!(
"Erasure set heal listing for bucket {bucket} is not advancing (repeated last version {page_last:?}); aborting to avoid an infinite loop"
)));
}
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(())
@@ -794,264 +798,6 @@ impl ErasureSetHealer {
progress.bytes_processed = 0; // set to 0 for now, can be extended later
progress.set_current_object(state.current_object.clone());
}
/// heal all buckets concurrently
#[allow(dead_code)]
async fn heal_buckets_concurrently(&self, buckets: &[String]) -> Vec<Result<()>> {
// use semaphore to control concurrency, avoid too many concurrent healings
let semaphore = Arc::new(tokio::sync::Semaphore::new(4)); // max 4 concurrent healings
let heal_futures = buckets.iter().map(|bucket| {
let bucket = bucket.clone();
let storage = self.storage.clone();
let progress = self.progress.clone();
let semaphore = semaphore.clone();
let cancel_token = self.cancel_token.clone();
async move {
let _permit = semaphore
.acquire()
.await
.map_err(|e| Error::other(format!("Failed to acquire semaphore for bucket heal: {e}")))?;
if cancel_token.is_cancelled() {
return Err(Error::TaskCancelled);
}
Self::heal_single_bucket(&storage, &bucket, &progress).await
}
});
// use join_all to process concurrently
join_all(heal_futures).await
}
/// heal single bucket
#[allow(dead_code)]
async fn heal_single_bucket(
storage: &Arc<dyn HealStorageAPI>,
bucket: &str,
progress: &Arc<RwLock<HealProgress>>,
) -> Result<()> {
debug!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
state = "started",
"Erasure set bucket started"
);
// 1. get bucket info
let _bucket_info = match storage.get_bucket_info(bucket).await? {
Some(info) => info,
None => {
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
state = "missing",
"Erasure set bucket heal skipped because bucket is missing"
);
return Ok(());
}
};
// 2. process objects with pagination to avoid loading all objects into memory
let mut continuation_token: Option<String> = None;
let mut total_scanned = 0u64;
let mut total_success = 0u64;
let mut total_failed = 0u64;
let heal_opts = HealOpts {
scan_mode: HealScanMode::Normal,
remove: true, // remove corrupted data
recreate: true, // recreate missing data
..Default::default()
};
loop {
// Get one page of objects
let (objects, next_token, is_truncated) = storage
.list_objects_for_heal_page(bucket, "", continuation_token.as_deref())
.await?;
let page_count = objects.len() as u64;
total_scanned += page_count;
// 3. update progress
{
let mut p = progress.write().await;
p.objects_scanned = total_scanned;
}
// 4. heal objects concurrently for this page
let object_results = Self::heal_objects_concurrently(storage, bucket, &objects, &heal_opts, progress).await;
// 5. count results for this page
let (success_count, failure_count) =
object_results
.into_iter()
.fold((0, 0), |(success, failure), result| match result {
Ok(_) => (success + 1, failure),
Err(_) => (success, failure + 1),
});
total_success += success_count;
total_failed += failure_count;
// 6. update progress
{
let mut p = progress.write().await;
p.objects_healed = total_success;
p.objects_failed = total_failed;
p.set_current_object(Some(format!("processing bucket: {bucket} (page)")));
}
// Check if there are more pages
if !is_truncated {
break;
}
continuation_token = next_heal_listing_token(bucket, "", next_token, is_truncated)?;
}
// 7. final progress update
{
let mut p = progress.write().await;
p.set_current_object(Some(format!("completed bucket: {bucket}")));
}
debug!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
total_success,
total_failed,
total_scanned,
state = "completed",
"Erasure set bucket completed"
);
Ok(())
}
/// heal objects concurrently
#[allow(dead_code)]
async fn heal_objects_concurrently(
storage: &Arc<dyn HealStorageAPI>,
bucket: &str,
objects: &[String],
heal_opts: &HealOpts,
_progress: &Arc<RwLock<HealProgress>>,
) -> Vec<Result<()>> {
// use semaphore to control object healing concurrency
let semaphore = Arc::new(tokio::sync::Semaphore::new(8)); // max 8 concurrent object healings
let heal_futures = objects.iter().map(|object| {
let object = object.clone();
let bucket = bucket.to_string();
let storage = storage.clone();
let heal_opts = *heal_opts;
let semaphore = semaphore.clone();
async move {
let _permit = semaphore
.acquire()
.await
.map_err(|e| Error::other(format!("Failed to acquire semaphore for object heal: {e}")))?;
match storage.heal_object(&bucket, &object, None, &heal_opts).await {
Ok((_result, None)) => {
debug!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
object = %object,
state = "healed",
"Erasure set object healed"
);
Ok(())
}
Ok((_, Some(err))) => {
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
object = %object,
state = "failed",
error = %err,
"Erasure set object heal failed"
);
Err(Error::other(err))
}
Err(err) => {
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
object = %object,
state = "failed",
error = %err,
"Erasure set object heal failed"
);
Err(err)
}
}
}
});
join_all(heal_futures).await
}
/// process results
#[allow(dead_code)]
async fn process_results(&self, results: Vec<Result<()>>) -> Result<()> {
let (success_count, failure_count): (usize, usize) =
results.into_iter().fold((0, 0), |(success, failure), result| match result {
Ok(_) => (success + 1, failure),
Err(_) => (success, failure + 1),
});
let total = success_count + failure_count;
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
success_count,
total,
state = "summary",
"Erasure set summary recorded"
);
if failure_count > 0 {
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
failure_count,
state = "summary_failed",
"Erasure set heal summary recorded failures"
);
return Err(Error::other(format!("{failure_count} buckets failed to heal")));
}
Ok(())
}
}
#[cfg(test)]
@@ -1190,3 +936,446 @@ mod tests {
));
}
}
#[cfg(test)]
mod resume_loop_tests {
//! Loop-level tests driving the private concurrent resume loop
//! (`heal_bucket_with_resume`) against a controllable fake `HealStorageAPI`
//! that emits programmable multi-version pages. These exercise the real loop
//! logic (cursor seeding, per-version dedup, anti-loop guard, absence
//! handling) — not merely a mock's own output.
use super::ErasureSetHealer;
use crate::heal::progress::HealProgress;
use crate::heal::resume::{CheckpointManager, ResumeManager, compose_key};
use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo, HealStorageAPI};
use crate::heal::storage_api::status::BucketInfo;
use crate::heal::{
BUCKET_META_PREFIX, DiskOption, DiskStore, EcstoreError, Endpoint, HealDiskExt as _, RUSTFS_META_BUCKET, new_disk,
};
use crate::{Error, Result};
use rustfs_common::heal_channel::{HealOpts, HealRequestSource};
use rustfs_madmin::heal_commands::HealResultItem;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use tempfile::TempDir;
use tokio::sync::RwLock;
use tokio_util::sync::CancellationToken;
fn item(name: &str, version: Option<&str>, delete_marker: bool) -> HealListItem {
HealListItem {
name: name.to_string(),
version_id: version.map(str::to_string),
is_delete_marker: delete_marker,
}
}
#[derive(Clone)]
struct Page {
items: Vec<HealListItem>,
next: Option<String>,
truncated: bool,
}
#[derive(Clone)]
enum HealOutcome {
Ok,
/// The version vanished before heal ran (deleted mid-heal).
VersionNotFound,
}
#[derive(Default)]
struct FakeStorage {
/// page keyed by the *incoming* continuation token
pages: Mutex<HashMap<Option<String>, Page>>,
/// per-`compose_key` heal outcome; default is `Ok`
outcomes: Mutex<HashMap<String, HealOutcome>>,
/// every heal_object call recorded as (name, version_id)
heal_calls: Mutex<Vec<(String, Option<String>)>>,
}
impl FakeStorage {
fn set_page(&self, token: Option<&str>, page: Page) {
self.pages.lock().unwrap().insert(token.map(str::to_string), page);
}
fn set_outcome(&self, name: &str, version: Option<&str>, outcome: HealOutcome) {
self.outcomes.lock().unwrap().insert(compose_key(name, version), outcome);
}
fn calls(&self) -> Vec<(String, Option<String>)> {
self.heal_calls.lock().unwrap().clone()
}
}
#[async_trait::async_trait]
impl HealStorageAPI for FakeStorage {
async fn get_object_meta(&self, _b: &str, _o: &str) -> Result<Option<HealObjectInfo>> {
Ok(None)
}
async fn get_object_data(&self, _b: &str, _o: &str) -> Result<Option<Vec<u8>>> {
Ok(None)
}
async fn put_object_data(&self, _b: &str, _o: &str, _d: &[u8]) -> Result<()> {
Ok(())
}
async fn delete_object(&self, _b: &str, _o: &str) -> Result<()> {
Ok(())
}
async fn verify_object_integrity(&self, _b: &str, _o: &str) -> Result<bool> {
Ok(true)
}
async fn ec_decode_rebuild(&self, _b: &str, _o: &str) -> Result<Vec<u8>> {
Ok(Vec::new())
}
async fn get_disk_status(&self, _e: &Endpoint) -> Result<DiskStatus> {
Ok(DiskStatus::Ok)
}
async fn format_disk(&self, _e: &Endpoint) -> Result<()> {
Ok(())
}
async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>> {
Ok(Some(BucketInfo {
name: bucket.to_string(),
..Default::default()
}))
}
async fn heal_bucket_metadata(&self, _b: &str) -> Result<()> {
Ok(())
}
async fn list_buckets(&self) -> Result<Vec<BucketInfo>> {
Ok(Vec::new())
}
async fn object_exists(&self, _b: &str, _o: &str) -> Result<bool> {
// Must never be consulted: the resume loop always goes through heal_object.
panic!("object_exists must not be called by the resume heal loop");
}
async fn get_object_size(&self, _b: &str, _o: &str) -> Result<Option<u64>> {
Ok(None)
}
async fn get_object_checksum(&self, _b: &str, _o: &str) -> Result<Option<String>> {
Ok(None)
}
async fn heal_object(
&self,
_bucket: &str,
object: &str,
version_id: Option<&str>,
_opts: &HealOpts,
) -> Result<(HealResultItem, Option<Error>)> {
self.heal_calls
.lock()
.unwrap()
.push((object.to_string(), version_id.map(str::to_string)));
let key = compose_key(object, version_id);
let outcome = self.outcomes.lock().unwrap().get(&key).cloned().unwrap_or(HealOutcome::Ok);
match outcome {
HealOutcome::Ok => Ok((HealResultItem::default(), None)),
HealOutcome::VersionNotFound => {
Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileVersionNotFound))))
}
}
}
async fn heal_bucket(&self, _b: &str, _o: &HealOpts) -> Result<HealResultItem> {
Ok(HealResultItem::default())
}
async fn heal_format(&self, _dry: bool) -> Result<(HealResultItem, Option<Error>)> {
Ok((HealResultItem::default(), None))
}
async fn list_objects_for_heal(&self, _b: &str, _p: &str) -> Result<Vec<HealListItem>> {
Ok(Vec::new())
}
async fn list_objects_for_heal_page(
&self,
_bucket: &str,
_prefix: &str,
continuation_token: Option<&str>,
) -> Result<(Vec<HealListItem>, Option<String>, bool)> {
let key = continuation_token.map(str::to_string);
let page = self.pages.lock().unwrap().get(&key).cloned();
match page {
Some(p) => Ok((p.items, p.next, p.truncated)),
None => Ok((Vec::new(), None, false)),
}
}
async fn get_disk_for_resume(&self, _id: &str) -> Result<DiskStore> {
Err(Error::other("not implemented in tests"))
}
}
async fn make_disk(temp: &TempDir) -> DiskStore {
let disk_path = temp.path().join("test_disk");
std::fs::create_dir_all(&disk_path).unwrap();
let endpoint = Endpoint::try_from(disk_path.to_string_lossy().as_ref()).unwrap();
let disk = new_disk(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.unwrap();
let _ = disk.make_volume(RUSTFS_META_BUCKET).await;
let _ = disk.make_volume(&format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}")).await;
disk
}
struct Env {
healer: ErasureSetHealer,
storage: Arc<FakeStorage>,
resume: ResumeManager,
checkpoint: CheckpointManager,
_temp: TempDir,
}
async fn make_env() -> Env {
let temp = TempDir::new().unwrap();
let disk = make_disk(&temp).await;
let storage = Arc::new(FakeStorage::default());
let healer = ErasureSetHealer::new(
storage.clone(),
Arc::new(RwLock::new(HealProgress::new())),
CancellationToken::new(),
disk.clone(),
HealOpts::default(),
HealRequestSource::Internal,
);
let resume = ResumeManager::new(
disk.clone(),
"task".to_string(),
"erasure_set".to_string(),
"pool_0_set_0".to_string(),
vec!["b".to_string()],
)
.await
.unwrap();
let checkpoint = CheckpointManager::new(disk, "task".to_string()).await.unwrap();
Env {
healer,
storage,
resume,
checkpoint,
_temp: temp,
}
}
/// Drive one bucket heal pass; returns (processed, successful, failed, skipped, result).
async fn run(env: &Env) -> (u64, u64, u64, u64, Result<()>) {
let mut current_object_index = 0usize;
let mut processed = 0u64;
let mut successful = 0u64;
let mut failed = 0u64;
let mut skipped = 0u64;
let result = env
.healer
.heal_bucket_with_resume(
"b",
"pool_0_set_0",
0,
&mut current_object_index,
&mut processed,
&mut successful,
&mut failed,
&mut skipped,
&env.resume,
&env.checkpoint,
)
.await;
(processed, successful, failed, skipped, result)
}
#[tokio::test]
async fn test_empty_bucket_no_panic() {
let env = make_env().await;
// no pages configured => empty, non-truncated page
let (processed, successful, failed, skipped, result) = run(&env).await;
result.expect("empty bucket must succeed");
assert_eq!(processed, 0);
assert_eq!(successful, 0);
assert_eq!(failed, 0);
assert_eq!(skipped, 0);
assert!(env.storage.calls().is_empty());
assert_eq!(env.resume.resume_cursor().await, None);
}
#[tokio::test]
async fn test_resume_across_page_boundary_no_drop_no_double() {
let env = make_env().await;
env.storage.set_page(
None,
Page {
items: vec![item("a", None, false), item("b", None, false)],
next: Some("t1".to_string()),
truncated: true,
},
);
env.storage.set_page(
Some("t1"),
Page {
items: vec![item("c", None, false), item("d", None, false)],
next: None,
truncated: false,
},
);
let (processed, successful, failed, _skipped, result) = run(&env).await;
result.expect("two-page heal must succeed");
assert_eq!(processed, 4);
assert_eq!(successful, 4);
assert_eq!(failed, 0);
let mut names: Vec<String> = env.storage.calls().into_iter().map(|(n, _)| n).collect();
names.sort();
assert_eq!(names, vec!["a", "b", "c", "d"], "every object exactly once, none dropped/doubled");
// Final page not truncated => cursor cleared.
assert_eq!(env.resume.resume_cursor().await, None);
}
#[tokio::test]
async fn test_object_with_versions_spanning_pages_advances() {
let env = make_env().await;
env.storage.set_page(
None,
Page {
items: vec![item("obj", Some("v1"), false)],
next: Some("t1".to_string()),
truncated: true,
},
);
env.storage.set_page(
Some("t1"),
Page {
items: vec![item("obj", Some("v2"), false)],
next: None,
truncated: false,
},
);
let (processed, _s, failed, _sk, result) = run(&env).await;
result.expect("object whose versions span pages must heal fully");
assert_eq!(processed, 2);
assert_eq!(failed, 0);
let calls = env.storage.calls();
assert!(calls.contains(&("obj".to_string(), Some("v1".to_string()))));
assert!(calls.contains(&("obj".to_string(), Some("v2".to_string()))));
}
#[tokio::test]
async fn test_non_advancing_cursor_aborts() {
let env = make_env().await;
// Both pages end on the same (name, version) even though the raw token
// advances (t1 -> t2): identity comparison must detect the stall.
env.storage.set_page(
None,
Page {
items: vec![item("a", None, false)],
next: Some("t1".to_string()),
truncated: true,
},
);
env.storage.set_page(
Some("t1"),
Page {
items: vec![item("a", None, false)],
next: Some("t2".to_string()),
truncated: true,
},
);
let (_p, _s, _f, _sk, result) = run(&env).await;
let err = result.expect_err("a non-advancing cursor must abort the loop");
assert!(err.to_string().contains("not advancing"), "unexpected error: {err}");
}
#[tokio::test]
async fn test_concurrent_page_dedup_exact_once() {
let env = make_env().await;
let items: Vec<HealListItem> = (0..50).map(|i| item(&format!("obj-{i}"), Some("v"), false)).collect();
env.storage.set_page(
None,
Page {
items,
next: None,
truncated: false,
},
);
let (processed, successful, failed, _sk, result) = run(&env).await;
result.expect("concurrent page must succeed");
assert_eq!(processed, 50);
assert_eq!(successful, 50);
assert_eq!(failed, 0);
let calls = env.storage.calls();
assert_eq!(calls.len(), 50, "each version healed exactly once under concurrency");
let unique: std::collections::HashSet<_> = calls.into_iter().collect();
assert_eq!(unique.len(), 50, "no version healed twice");
}
#[tokio::test]
async fn test_resume_after_version_deleted_midheal_no_skip() {
let env = make_env().await;
// Simulate a resumed pass: (a,v1) was already processed last time, and
// the cursor points at the in-flight page.
env.checkpoint
.add_processed_object(compose_key("a", Some("v1")))
.await
.unwrap();
env.resume.set_resume_cursor(Some("t0".to_string())).await.unwrap();
env.storage.set_page(
Some("t0"),
Page {
items: vec![
item("a", Some("v1"), false), // already done -> deduped
item("a", Some("v2"), false), // deleted mid-heal
item("c", None, true), // delete-marker latest, still healed
],
next: None,
truncated: false,
},
);
// v2 vanished before heal ran.
env.storage.set_outcome("a", Some("v2"), HealOutcome::VersionNotFound);
let (_p, _s, failed, skipped, result) = run(&env).await;
result.expect("resume after mid-heal deletion must succeed");
// Genuine absence is handled (Ok), never counted as a failure.
assert_eq!(failed, 0, "a deleted version must not be a failure");
assert_eq!(skipped, 0, "absence is treated as healed-ok, not skipped");
let calls = env.storage.calls();
assert!(
!calls.contains(&("a".to_string(), Some("v1".to_string()))),
"already-processed version must be deduped, not re-healed"
);
assert!(
calls.contains(&("a".to_string(), Some("v2".to_string()))),
"the surviving-but-now-gone version must still be attempted, not skipped"
);
assert!(calls.contains(&("c".to_string(), None)), "delete-marker latest must be healed");
}
#[tokio::test]
async fn test_schedule_retry_resets_both_managers_and_reheals() {
let env = make_env().await;
// Seed some progress that a retry must discard.
env.checkpoint
.add_processed_object(compose_key("a", Some("v1")))
.await
.unwrap();
env.checkpoint.update_position(1, 42).await.unwrap();
env.resume.set_resume_cursor(Some("t9".to_string())).await.unwrap();
assert_eq!(env.resume.resume_cursor().await, Some("t9".to_string()));
// The retry branch calls BOTH together.
assert!(env.resume.schedule_retry().await.unwrap(), "retry budget should be available");
env.checkpoint.reset_for_retry().await.unwrap();
// Resume state: cursor cleared, retry counted, progress zeroed.
let state = env.resume.get_state().await;
assert_eq!(env.resume.resume_cursor().await, None, "cursor must be cleared for a full re-scan");
assert_eq!(state.retry_count, 1);
// Checkpoint: dedup sets and position cleared so the retry re-heals everything.
let checkpoint = env.checkpoint.get_checkpoint().await;
assert!(checkpoint.processed_objects.is_empty());
assert_eq!(checkpoint.current_object_index, 0);
}
}
crates/heal/src/heal/manager.rs
+2 -2
View File
@@ -3039,7 +3039,7 @@ mod tests {
Ok((HealResultItem::default(), None))
}
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<String>> {
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<crate::heal::storage::HealListItem>> {
Ok(Vec::new())
}
@@ -3048,7 +3048,7 @@ mod tests {
_bucket: &str,
_prefix: &str,
_continuation_token: Option<&str>,
) -> Result<(Vec<String>, Option<String>, bool)> {
) -> Result<(Vec<crate::heal::storage::HealListItem>, Option<String>, bool)> {
Ok((Vec::new(), None, false))
}
crates/heal/src/heal/resume.rs
+247 -4
View File
@@ -34,6 +34,25 @@ const RESUME_STATE_FILE: &str = "ahm_resume_state.json";
const RESUME_PROGRESS_FILE: &str = "ahm_progress.json";
const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json";
/// Current on-disk schema version for `ResumeState`. Snapshots written by an
/// older schema (which tracked latest-only object names and a positional
/// cursor) are incompatible with the per-version resume cursor, so they are
/// discarded on load and the scan restarts from the beginning.
const CURRENT_RESUME_SCHEMA: u32 = 2;
/// Current on-disk schema version for `ResumeCheckpoint`. Same rationale as
/// `CURRENT_RESUME_SCHEMA`: pre-per-version dedup identities are not comparable
/// to the new `compose_key` identities, so a stale checkpoint is discarded.
const CURRENT_CHECKPOINT_SCHEMA: u32 = 2;
/// Build the canonical, provably-injective dedup identity for an object
/// version. Length-prefixing the object key makes the encoding injective: no
/// two distinct `(object, version_id)` pairs can collide, even for adversarial
/// keys containing `:` or embedded null bytes. This is the single source of
/// truth for per-version dedup across the heal loop and the checkpoint sets.
pub fn compose_key(object: &str, version_id: Option<&str>) -> String {
format!("{}:{}{}", object.len(), object, version_id.unwrap_or(""))
}
/// 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.
@@ -76,6 +95,13 @@ fn path_to_str(path: &Path) -> Result<&str> {
/// resume state
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResumeState {
/// on-disk schema version; absent in legacy snapshots (defaults to 0)
#[serde(default)]
pub schema_version: u32,
/// authoritative opaque `(marker, version_marker)` continuation token for
/// the version listing. `None` means "start from the beginning".
#[serde(default)]
pub resume_cursor: Option<String>,
/// task id
pub task_id: String,
/// task type
@@ -118,6 +144,8 @@ pub struct ResumeState {
impl ResumeState {
pub fn new(task_id: String, task_type: String, set_disk_id: String, buckets: Vec<String>) -> Self {
Self {
schema_version: CURRENT_RESUME_SCHEMA,
resume_cursor: None,
task_id,
task_type,
set_disk_id,
@@ -153,6 +181,18 @@ impl ResumeState {
self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
}
/// Read the authoritative version-listing continuation cursor.
pub fn resume_cursor(&self) -> Option<String> {
self.resume_cursor.clone()
}
/// Persist the authoritative version-listing continuation cursor. `None`
/// resets the scan to the beginning of the current bucket.
pub fn set_resume_cursor(&mut self, cursor: Option<String>) {
self.resume_cursor = cursor;
self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
}
pub fn complete_bucket(&mut self, bucket: &str) {
if !self.completed_buckets.contains(&bucket.to_string()) {
self.completed_buckets.push(bucket.to_string());
@@ -178,6 +218,9 @@ impl ResumeState {
self.failed_objects = 0;
self.skipped_objects = 0;
self.completed = false;
// A retry re-scans every bucket from the beginning, so the version
// cursor must be cleared too — otherwise the retry would resume mid-scan.
self.resume_cursor = None;
self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
}
@@ -252,10 +295,34 @@ impl ResumeManager {
/// load resume state from disk
pub async fn load_from_disk(disk: DiskStore, task_id: &str) -> Result<Self> {
let state_data = Self::read_state_file(&disk, task_id).await?;
let state: ResumeState = serde_json::from_slice(&state_data).map_err(|e| Error::TaskExecutionFailed {
let mut state: ResumeState = serde_json::from_slice(&state_data).map_err(|e| Error::TaskExecutionFailed {
message: format!("Failed to deserialize resume state: {e}"),
})?;
// A snapshot written by an older schema tracked a latest-only positional
// cursor that is meaningless under per-version resume. Discard the stale
// progress so the scan restarts cleanly, then stamp the current schema.
if state.schema_version < CURRENT_RESUME_SCHEMA {
warn!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
task_id,
found_schema = state.schema_version,
current_schema = CURRENT_RESUME_SCHEMA,
state = "schema_discarded",
"Heal resume state schema is stale; discarding cursor and progress"
);
state.resume_cursor = None;
state.processed_objects = 0;
state.successful_objects = 0;
state.failed_objects = 0;
state.skipped_objects = 0;
state.completed = false;
state.schema_version = CURRENT_RESUME_SCHEMA;
}
Ok(Self {
disk,
state: Arc::new(RwLock::new(state)),
@@ -312,6 +379,21 @@ impl ResumeManager {
result
}
/// Read the authoritative version-listing continuation cursor.
pub async fn resume_cursor(&self) -> Option<String> {
self.state.read().await.resume_cursor()
}
/// Persist the authoritative version-listing continuation cursor. This is
/// written unthrottled (once per completed page) so a crash always resumes
/// from a real page boundary.
pub async fn set_resume_cursor(&self, cursor: Option<String>) -> Result<()> {
let mut state = self.state.write().await;
state.set_resume_cursor(cursor);
drop(state);
self.save_state().await
}
/// complete bucket
pub async fn complete_bucket(&self, bucket: &str) -> Result<()> {
let mut state = self.state.write().await;
@@ -450,6 +532,9 @@ impl ResumeManager {
/// resume checkpoint
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResumeCheckpoint {
/// on-disk schema version; absent in legacy snapshots (defaults to 0)
#[serde(default)]
pub schema_version: u32,
/// task id
pub task_id: String,
/// checkpoint time
@@ -472,6 +557,7 @@ pub struct ResumeCheckpoint {
impl ResumeCheckpoint {
pub fn new(task_id: String) -> Self {
Self {
schema_version: CURRENT_CHECKPOINT_SCHEMA,
task_id,
checkpoint_time: SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(),
current_bucket_index: 0,
@@ -555,9 +641,33 @@ impl CheckpointManager {
/// load checkpoint from disk
pub async fn load_from_disk(disk: DiskStore, task_id: &str) -> Result<Self> {
let checkpoint_data = Self::read_checkpoint_file(&disk, task_id).await?;
let checkpoint: ResumeCheckpoint = serde_json::from_slice(&checkpoint_data).map_err(|e| Error::TaskExecutionFailed {
message: format!("Failed to deserialize checkpoint: {e}"),
})?;
let mut checkpoint: ResumeCheckpoint =
serde_json::from_slice(&checkpoint_data).map_err(|e| Error::TaskExecutionFailed {
message: format!("Failed to deserialize checkpoint: {e}"),
})?;
// A checkpoint from an older schema stored latest-only dedup identities
// that are not comparable to the new per-version `compose_key`
// identities. Discard the stale sets and position, then stamp the
// current schema so the scan restarts cleanly.
if checkpoint.schema_version < CURRENT_CHECKPOINT_SCHEMA {
warn!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_CHECKPOINT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
task_id,
found_schema = checkpoint.schema_version,
current_schema = CURRENT_CHECKPOINT_SCHEMA,
state = "schema_discarded",
"Heal checkpoint schema is stale; discarding dedup sets and position"
);
checkpoint.processed_objects.clear();
checkpoint.failed_objects.clear();
checkpoint.skipped_objects.clear();
checkpoint.current_object_index = 0;
checkpoint.schema_version = CURRENT_CHECKPOINT_SCHEMA;
}
Ok(Self {
disk,
@@ -957,6 +1067,139 @@ mod tests {
assert!(checkpoint.skipped_objects.contains("c"));
}
#[test]
fn test_compose_key_injective_with_adversarial_keys() {
// Length-prefixing must keep the encoding injective even when keys
// contain the delimiter, embedded nulls, or look like a composed key.
assert_ne!(compose_key("a\0b", None), compose_key("a", Some("b")));
assert_ne!(compose_key("3:xy", None), compose_key("x", Some("y")));
assert_ne!(compose_key("a:b", None), compose_key("a", Some("b")));
assert_ne!(compose_key("", Some("x")), compose_key("x", None));
// Identical inputs must produce identical keys (stable identity).
assert_eq!(compose_key("obj", Some("v1")), compose_key("obj", Some("v1")));
}
#[test]
fn test_composite_key_dedup_distinguishes_versions() {
// Two versions of the same object must be distinct dedup identities, and
// the delete-marker/nil (None) version must not collide with a real one.
let mut checkpoint = ResumeCheckpoint::new("task".to_string());
checkpoint.add_processed_object(compose_key("obj", Some("v1")));
checkpoint.add_processed_object(compose_key("obj", Some("v2")));
checkpoint.add_processed_object(compose_key("obj", None));
assert_eq!(checkpoint.processed_objects.len(), 3);
assert!(checkpoint.processed_objects.contains(&compose_key("obj", Some("v1"))));
assert!(checkpoint.processed_objects.contains(&compose_key("obj", Some("v2"))));
assert!(checkpoint.processed_objects.contains(&compose_key("obj", None)));
// A different object with the same version id is still distinct.
assert!(!checkpoint.processed_objects.contains(&compose_key("other", Some("v1"))));
}
#[tokio::test]
async fn test_resumestate_schema_v0_discarded_on_load() {
use super::super::{DiskOption, Endpoint, new_disk};
use tempfile::TempDir;
let temp_dir = TempDir::new().unwrap();
let disk_path = temp_dir.path().join("test_disk");
std::fs::create_dir_all(&disk_path).unwrap();
let endpoint = Endpoint::try_from(disk_path.to_string_lossy().as_ref()).unwrap();
let disk = new_disk(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.unwrap();
let _ = disk.make_volume(RUSTFS_META_BUCKET).await;
let _ = disk.make_volume(&format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}")).await;
// Legacy snapshot: no schema_version, a stale positional cursor and progress.
let legacy = r#"{
"task_id": "old-task",
"task_type": "erasure_set",
"set_disk_id": "pool_0_set_0",
"start_time": 1700000000,
"last_update": 1700000000,
"completed": true,
"total_objects": 100,
"processed_objects": 50,
"successful_objects": 40,
"failed_objects": 10,
"skipped_objects": 0,
"current_bucket": null,
"current_object": null,
"completed_buckets": ["b1"],
"pending_buckets": [],
"error_message": null,
"retry_count": 1,
"max_retries": 3,
"resume_cursor": "v1:stale-token"
}"#;
let file_path = format!("{BUCKET_META_PREFIX}/old-task_{RESUME_STATE_FILE}");
disk.write_all(RUSTFS_META_BUCKET, &file_path, legacy.as_bytes().to_vec().into())
.await
.unwrap();
let manager = ResumeManager::load_from_disk(disk.clone(), "old-task").await.unwrap();
let state = manager.get_state().await;
assert_eq!(state.schema_version, CURRENT_RESUME_SCHEMA, "schema must be stamped current");
assert_eq!(state.resume_cursor, None, "stale cursor must be cleared");
assert_eq!(state.processed_objects, 0);
assert_eq!(state.successful_objects, 0);
assert_eq!(state.failed_objects, 0);
assert!(!state.completed);
temp_dir.close().unwrap();
}
#[tokio::test]
async fn test_checkpoint_schema_v0_discarded_on_load() {
use super::super::{DiskOption, Endpoint, new_disk};
use tempfile::TempDir;
let temp_dir = TempDir::new().unwrap();
let disk_path = temp_dir.path().join("test_disk");
std::fs::create_dir_all(&disk_path).unwrap();
let endpoint = Endpoint::try_from(disk_path.to_string_lossy().as_ref()).unwrap();
let disk = new_disk(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.unwrap();
let _ = disk.make_volume(RUSTFS_META_BUCKET).await;
let _ = disk.make_volume(&format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}")).await;
// Legacy checkpoint: no schema_version, stale position and dedup sets.
let legacy = r#"{
"task_id": "old-task",
"checkpoint_time": 1700000000,
"current_bucket_index": 2,
"current_object_index": 500,
"processed_objects": ["a", "b"],
"failed_objects": ["c"],
"skipped_objects": ["d"]
}"#;
let file_path = format!("{BUCKET_META_PREFIX}/old-task_{RESUME_CHECKPOINT_FILE}");
disk.write_all(RUSTFS_META_BUCKET, &file_path, legacy.as_bytes().to_vec().into())
.await
.unwrap();
let manager = CheckpointManager::load_from_disk(disk.clone(), "old-task").await.unwrap();
let checkpoint = manager.get_checkpoint().await;
assert_eq!(checkpoint.schema_version, CURRENT_CHECKPOINT_SCHEMA, "schema must be stamped current");
assert_eq!(checkpoint.current_object_index, 0, "stale position must be reset");
assert!(checkpoint.processed_objects.is_empty());
assert!(checkpoint.failed_objects.is_empty());
assert!(checkpoint.skipped_objects.is_empty());
temp_dir.close().unwrap();
}
#[test]
fn test_persist_throttle_batches_until_threshold() {
let mut throttle = PersistThrottle::new();
crates/heal/src/heal/storage.rs
+251 -31
View File
@@ -14,8 +14,11 @@
use crate::{Error, Result};
use async_trait::async_trait;
use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use rustfs_common::heal_channel::{HealOpts, HealScanMode};
use rustfs_madmin::heal_commands::HealResultItem;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tracing::{debug, error, warn};
@@ -44,9 +47,156 @@ pub(crate) fn next_heal_listing_token(
return Ok(None);
}
next_token.map(Some).ok_or_else(|| Error::TaskExecutionFailed {
message: format!("Object listing for {bucket}/{prefix} was truncated without continuation token"),
})
match next_token {
Some(token) => Ok(Some(token)),
None => {
// A version listing legitimately reports the final page as truncated
// when the last object's versions land exactly on the page boundary
// yet the backend has nothing further to yield. Treat a missing
// continuation token as end-of-listing rather than a hard error so
// the heal pass terminates cleanly instead of failing the bucket.
warn!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "next_heal_listing_token",
bucket,
prefix,
state = "truncated_without_token",
"Heal storage object listing truncated without continuation token; treating as end of listing"
);
Ok(None)
}
}
}
/// Opaque continuation token payload for heal version listing. Encodes the
/// `(marker, version_marker)` pair that `list_object_versions` needs to resume.
#[derive(Debug, Serialize, Deserialize)]
struct HealTokenPayload {
/// object key marker
#[serde(rename = "m")]
m: Option<String>,
/// version-id marker
#[serde(rename = "v")]
v: Option<String>,
}
const HEAL_TOKEN_PREFIX: &str = "v1:";
/// Encode a `(marker, version_marker)` pair into an opaque heal continuation
/// token. The token is `"v1:" + base64url_nopad(json)`.
///
/// Invariant: `list_object_versions` returns `NotImplemented` for
/// `(None, Some(_))`, so callers must never produce that pair. This is checked
/// with a `debug_assert!`.
pub(crate) fn encode_heal_token(marker: Option<&str>, version_marker: Option<&str>) -> String {
debug_assert!(
!(marker.is_none() && version_marker.is_some()),
"encode_heal_token must never be called with (None, Some(_))"
);
let payload = HealTokenPayload {
m: marker.map(str::to_string),
v: version_marker.map(str::to_string),
};
// serde_json of a simple two-Option struct cannot fail; fall back to an
// empty object rather than panicking if it somehow does.
let json = serde_json::to_vec(&payload).unwrap_or_else(|_| b"{}".to_vec());
format!("{HEAL_TOKEN_PREFIX}{}", URL_SAFE_NO_PAD.encode(json))
}
/// Decode an opaque heal continuation token back into `(marker, version_marker)`.
///
/// TOTAL function: an empty token, a missing `"v1:"` prefix, invalid base64, or
/// invalid JSON all decode to `(None, None)` (start from the beginning). A
/// decoded `(None, Some(_))` is coerced to `(None, None)` to preserve the
/// `list_object_versions` invariant.
pub(crate) fn decode_heal_token(token: &str) -> (Option<String>, Option<String>) {
if token.is_empty() {
return (None, None);
}
let Some(encoded) = token.strip_prefix(HEAL_TOKEN_PREFIX) else {
warn!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "decode_heal_token",
state = "missing_prefix",
"Heal continuation token missing version prefix; restarting listing"
);
return (None, None);
};
let bytes = match URL_SAFE_NO_PAD.decode(encoded) {
Ok(bytes) => bytes,
Err(e) => {
warn!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "decode_heal_token",
state = "bad_base64",
error = %e,
"Heal continuation token has invalid base64; restarting listing"
);
return (None, None);
}
};
let payload: HealTokenPayload = match serde_json::from_slice(&bytes) {
Ok(payload) => payload,
Err(e) => {
warn!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "decode_heal_token",
state = "bad_json",
error = %e,
"Heal continuation token has invalid payload; restarting listing"
);
return (None, None);
}
};
// Preserve the list_object_versions invariant: (None, Some(_)) is illegal.
if payload.m.is_none() && payload.v.is_some() {
warn!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "decode_heal_token",
state = "illegal_version_only_marker",
"Heal continuation token had a version marker without an object marker; restarting listing"
);
return (None, None);
}
(payload.m, payload.v)
}
/// A single object version to heal.
///
/// `is_delete_marker` is OBSERVABILITY-ONLY (metrics / logging / e2e
/// assertions); it MUST NOT gate healing logic. Whether the delete-marker path
/// or the data path is taken is decided internally in `ops/heal.rs` from
/// `latest_meta.deleted`. `version_id` is normalized (nil/absent UUID => `None`)
/// at the single construction point in `list_objects_for_heal_page`.
#[derive(Debug, Clone)]
pub struct HealListItem {
/// object key
pub name: String,
/// normalized version id (`None` when the version is nil/absent)
pub version_id: Option<String>,
/// whether this version is a delete marker (observability only)
pub is_delete_marker: bool,
}
/// Disk status for heal operations
@@ -132,20 +282,21 @@ pub trait HealStorageAPI: Send + Sync {
/// Heal format using ecstore
async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)>;
/// List objects for healing (returns all objects, may use significant memory for large buckets)
/// List object versions for healing (returns all versions, may use significant memory for large buckets)
///
/// WARNING: This method loads all objects into memory at once. For buckets with many objects,
/// consider using `list_objects_for_heal_page` instead to process objects in pages.
async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result<Vec<String>>;
/// WARNING: This method loads all object versions into memory at once. For buckets with many
/// objects/versions, consider using `list_objects_for_heal_page` instead to process versions in pages.
async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result<Vec<HealListItem>>;
/// List objects for healing with pagination (returns one page and continuation token)
/// Returns (objects, next_continuation_token, is_truncated)
/// List object versions for healing with pagination (returns one page and continuation token)
/// Returns (versions, next_continuation_token, is_truncated). The continuation token is an
/// opaque composite `(marker, version_marker)` value — see `encode_heal_token`/`decode_heal_token`.
async fn list_objects_for_heal_page(
&self,
bucket: &str,
prefix: &str,
continuation_token: Option<&str>,
) -> Result<(Vec<String>, Option<String>, bool)>;
) -> Result<(Vec<HealListItem>, Option<String>, bool)>;
/// Get disk for resume functionality
async fn get_disk_for_resume(&self, set_disk_id: &str) -> Result<DiskStore>;
@@ -1063,7 +1214,7 @@ impl HealStorageAPI for ECStoreHealStorage {
}
}
async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result<Vec<String>> {
async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result<Vec<HealListItem>> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
@@ -1084,10 +1235,10 @@ impl HealStorageAPI for ECStoreHealStorage {
bucket,
prefix,
state = "memory_heavy",
"Heal storage object listing loads all objects into memory"
"Heal storage version listing loads all versions into memory (footprint is per-version, not per-object)"
);
let mut all_objects = Vec::new();
let mut all_objects: Vec<HealListItem> = Vec::new();
let mut continuation_token: Option<String> = None;
loop {
@@ -1102,6 +1253,9 @@ impl HealStorageAPI for ECStoreHealStorage {
}
continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
if continuation_token.is_none() {
break;
}
}
debug!(
@@ -1124,7 +1278,7 @@ impl HealStorageAPI for ECStoreHealStorage {
bucket: &str,
prefix: &str,
continuation_token: Option<&str>,
) -> Result<(Vec<String>, Option<String>, bool)> {
) -> Result<(Vec<HealListItem>, Option<String>, bool)> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
@@ -1139,13 +1293,17 @@ impl HealStorageAPI for ECStoreHealStorage {
);
const MAX_KEYS: i32 = 1000;
let continuation_token_opt = continuation_token.map(|s| s.to_string());
// Decode the opaque composite token into the (marker, version_marker)
// pair that list_object_versions consumes. Malformed tokens restart the
// listing from the beginning (decode_heal_token is total).
let (marker, version_marker) = decode_heal_token(continuation_token.unwrap_or(""));
// Use list_objects_v2 to get objects with pagination
// Enumerate EVERY version (not just the latest) so old versions and
// delete-marker-latest objects are healed too.
let list_info = match self
.ecstore
.clone()
.list_objects_v2(bucket, prefix, continuation_token_opt, None, MAX_KEYS, false, None, false)
.list_object_versions(bucket, prefix, marker, version_marker, None, MAX_KEYS)
.await
{
Ok(info) => info,
@@ -1166,25 +1324,43 @@ impl HealStorageAPI for ECStoreHealStorage {
}
};
// Collect objects from this page
let page_objects: Vec<String> = list_info.objects.into_iter().map(|obj| obj.name).collect();
// Collect versions from this page. version_id is normalized to Option<String>
// here at the single construction point: nil/absent UUID => None.
let page_objects: Vec<HealListItem> = list_info
.objects
.into_iter()
.map(|obj| HealListItem {
name: obj.name,
version_id: obj.version_id.filter(|u| !u.is_nil()).map(|u| u.to_string()),
is_delete_marker: obj.delete_marker,
})
.collect();
let page_count = page_objects.len();
let next_token = if list_info.is_truncated {
Some(encode_heal_token(
list_info.next_marker.as_deref(),
list_info.next_version_idmarker.as_deref(),
))
} else {
None
};
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "list_objects_for_heal_page",
bucket,
prefix,
object_count = page_count,
is_truncated = list_info.is_truncated,
state = "page_loaded",
"Heal storage object listing page loaded"
bucket,
prefix,
version_count = page_count,
is_truncated = list_info.is_truncated,
state = "page_loaded",
"Heal storage version listing page loaded"
);
Ok((page_objects, list_info.next_continuation_token, list_info.is_truncated))
Ok((page_objects, next_token, list_info.is_truncated))
}
async fn get_disk_for_resume(&self, set_disk_id: &str) -> Result<DiskStore> {
@@ -1234,7 +1410,11 @@ impl HealStorageAPI for ECStoreHealStorage {
#[cfg(test)]
mod tests {
use super::super::StorageError;
use super::{is_transient_object_exists_error, is_transient_object_exists_message, next_heal_listing_token};
use super::{
decode_heal_token, encode_heal_token, is_transient_object_exists_error, is_transient_object_exists_message,
next_heal_listing_token,
};
use base64::Engine as _;
#[test]
fn next_heal_listing_token_returns_none_for_complete_page() {
@@ -1254,11 +1434,51 @@ mod tests {
}
#[test]
fn next_heal_listing_token_fails_for_truncated_page_without_token() {
let err = next_heal_listing_token("bucket", "prefix", None, true).expect_err("truncated page without token must fail");
fn next_heal_listing_token_treats_truncated_page_without_token_as_end() {
// A version listing can report the final page as truncated with no
// continuation token; that must terminate the scan cleanly, not error.
assert_eq!(
next_heal_listing_token("bucket", "prefix", None, true).expect("truncated without token ends listing"),
None
);
}
assert!(matches!(err, super::Error::TaskExecutionFailed { .. }));
assert!(err.to_string().contains("truncated without continuation token"));
#[test]
fn test_heal_token_roundtrip() {
let token = encode_heal_token(Some("obj/key"), Some("v-123"));
assert!(token.starts_with("v1:"));
assert_eq!(decode_heal_token(&token), (Some("obj/key".to_string()), Some("v-123".to_string())));
// marker only (no version marker) round-trips.
let token = encode_heal_token(Some("obj/key"), None);
assert_eq!(decode_heal_token(&token), (Some("obj/key".to_string()), None));
// (None, None) round-trips.
let token = encode_heal_token(None, None);
assert_eq!(decode_heal_token(&token), (None, None));
}
#[test]
fn test_heal_token_malformed_resets_to_start() {
// empty, wrong prefix, bad base64, and bad json all reset to (None, None).
assert_eq!(decode_heal_token(""), (None, None));
assert_eq!(decode_heal_token("no-prefix-here"), (None, None));
assert_eq!(decode_heal_token("v1:!!!not-base64!!!"), (None, None));
// valid base64 of non-JSON bytes.
let bad_json = format!("v1:{}", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b"not json"));
assert_eq!(decode_heal_token(&bad_json), (None, None));
// a raw v2-style list_objects_v2 token (no "v1:" prefix) resets cleanly.
assert_eq!(decode_heal_token("some-opaque-legacy-token"), (None, None));
}
#[test]
fn test_heal_token_none_and_marker_only() {
// A decoded payload must NEVER yield (None, Some(_)) because
// list_object_versions returns NotImplemented for that pairing.
// Craft a token whose JSON encodes (None, Some) directly and confirm coercion.
let json = br#"{"m":null,"v":"orphan-version"}"#;
let token = format!("v1:{}", base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json));
assert_eq!(decode_heal_token(&token), (None, None), "version-only marker must coerce to (None, None)");
}
#[test]
crates/heal/src/heal/task.rs
+116 -115
View File
@@ -1315,8 +1315,9 @@ impl HealTask {
)
.await?;
for object in objects {
for item in objects {
self.check_control_flags().await?;
let object = item.name.as_str();
scanned += 1;
{
let mut progress = self.progress.write().await;
@@ -1324,109 +1325,95 @@ impl HealTask {
progress.update_progress(scanned, healed, failed, bytes);
}
match self.await_with_control(self.storage.object_exists(bucket, &object)).await {
Ok(false) => {
// Heal each enumerated version directly. There is no latest-only
// existence gate: genuine absence surfaces through heal_object's
// not-found result and is handled below.
match self
.await_with_control(
self.storage
.heal_object(bucket, object, item.version_id.as_deref(), &heal_opts),
)
.await
{
Ok((result, None)) => {
healed += 1;
bytes = bytes.saturating_add(result.object_size as u64);
self.record_result_item(result).await;
}
Ok(true) => match self
.await_with_control(self.storage.heal_object(bucket, &object, None, &heal_opts))
.await
{
Ok((result, None)) => {
Ok((_, Some(err))) => {
if is_missing_object_dir_heal_result(object, &err) {
healed += 1;
bytes = bytes.saturating_add(result.object_size as u64);
self.record_result_item(result).await;
debug!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object = %object,
result = "object_dir_not_found_skipped",
"Heal bucket object-dir candidate skipped after not-found result"
);
continue;
}
Ok((_, Some(err))) => {
if is_missing_object_dir_heal_result(&object, &err) {
healed += 1;
debug!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object = %object,
result = "object_dir_not_found_skipped",
"Heal bucket object-dir candidate skipped after not-found result"
);
continue;
}
if Self::should_skip_data_usage_cache_heal_error(bucket, &object, &err) {
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object = %object,
result = "transient_skip",
error = %err,
"Heal bucket object repair skipped due to transient error"
);
} else {
failed += 1;
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object = %object,
result = "object_failed",
error = %err,
"Heal bucket object repair failed"
);
}
if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) {
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object = %object,
result = "transient_skip",
error = %err,
"Heal bucket object repair skipped due to transient error"
);
} else {
failed += 1;
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object = %object,
result = "object_failed",
error = %err,
"Heal bucket object repair failed"
);
}
Err(err) => {
if Self::should_skip_data_usage_cache_heal_error(bucket, &object, &err) {
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object = %object,
result = "transient_skip",
error = %err,
"Heal bucket object repair skipped due to transient error"
);
} else {
failed += 1;
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object = %object,
result = "object_failed",
error = %err,
"Heal bucket object repair failed"
);
}
}
},
}
Err(err) => {
failed += 1;
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object = %object,
result = "precheck_failed",
error = %err,
"Heal bucket object precheck failed"
);
if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) {
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object = %object,
result = "transient_skip",
error = %err,
"Heal bucket object repair skipped due to transient error"
);
} else {
failed += 1;
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
object = %object,
result = "object_failed",
error = %err,
"Heal bucket object repair failed"
);
}
}
}
@@ -1441,6 +1428,10 @@ impl HealTask {
}
continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
if continuation_token.is_none() {
// Truncated but no continuation token: end of listing.
break;
}
}
if failed > 0 {
@@ -2230,7 +2221,7 @@ impl std::fmt::Debug for HealTask {
mod tests {
use super::super::{DiskStore, Endpoint};
use super::*;
use crate::heal::storage::{DiskStatus, HealObjectInfo};
use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo};
use rustfs_madmin::heal_commands::HealResultItem;
use std::collections::HashMap;
use std::sync::Mutex;
@@ -2254,6 +2245,15 @@ mod tests {
include_object_dir_candidate: Mutex<bool>,
}
/// Build a latest, non-delete-marker heal list item with no version id.
fn heal_item(name: &str) -> HealListItem {
HealListItem {
name: name.to_string(),
version_id: None,
is_delete_marker: false,
}
}
enum MockHealObjectOutcome {
OkWithOtherError(&'static str),
ErrOther(&'static str),
@@ -2407,8 +2407,8 @@ mod tests {
}
}
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<String>> {
Ok(vec!["object-a".to_string(), "object-b".to_string()])
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<HealListItem>> {
Ok(vec![heal_item("object-a"), heal_item("object-b")])
}
async fn list_objects_for_heal_page(
@@ -2416,10 +2416,10 @@ mod tests {
bucket: &str,
prefix: &str,
continuation_token: Option<&str>,
) -> Result<(Vec<String>, Option<String>, bool)> {
) -> Result<(Vec<HealListItem>, Option<String>, bool)> {
self.listed_prefixes.lock().unwrap().push(prefix.to_string());
if *self.truncate_without_token.lock().unwrap() {
return Ok((vec!["object-a".to_string()], None, true));
return Ok((vec![heal_item("object-a")], None, true));
}
let mut listed = self.listed.lock().unwrap();
@@ -2427,15 +2427,15 @@ mod tests {
*listed = true;
let objects = if bucket == RUSTFS_META_BUCKET {
vec![
format!("{BUCKET_META_PREFIX}/{DATA_USAGE_CACHE_NAME}"),
format!("{BUCKET_META_PREFIX}/bucket-metadata.bin"),
heal_item(&format!("{BUCKET_META_PREFIX}/{DATA_USAGE_CACHE_NAME}")),
heal_item(&format!("{BUCKET_META_PREFIX}/bucket-metadata.bin")),
]
} else if prefix == "logs/" {
vec!["logs/object-a".to_string(), "logs/object-b".to_string()]
vec![heal_item("logs/object-a"), heal_item("logs/object-b")]
} else if *self.include_object_dir_candidate.lock().unwrap() {
vec!["object-a".to_string(), "object-dir/".to_string(), "object-b".to_string()]
vec![heal_item("object-a"), heal_item("object-dir/"), heal_item("object-b")]
} else {
vec!["object-a".to_string(), "object-b".to_string()]
vec![heal_item("object-a"), heal_item("object-b")]
};
Ok((objects, None, false))
} else {
@@ -2514,7 +2514,11 @@ mod tests {
}
#[tokio::test]
async fn test_recursive_bucket_heal_fails_when_listing_lacks_continuation_token() {
async fn test_recursive_bucket_heal_treats_missing_continuation_token_as_end() {
// A version listing can report the final page as truncated with no
// continuation token. That is treated as end-of-listing (not an error),
// so the returned page is healed and the pass terminates cleanly instead
// of erroring or looping forever.
let storage = Arc::new(MockStorage {
truncate_without_token: Mutex::new(true),
..Default::default()
@@ -2532,17 +2536,14 @@ mod tests {
);
let task = HealTask::from_request(request, storage.clone());
let err = task
.heal_bucket("bucket-a")
task.heal_bucket("bucket-a")
.await
.expect_err("recursive bucket heal must fail on incomplete pagination state");
.expect("truncated-without-token must terminate cleanly, not loop or error");
assert!(matches!(err, Error::TaskExecutionFailed { .. }));
assert!(err.to_string().contains("truncated without continuation token"));
assert_eq!(
storage.healed_objects.lock().unwrap().as_slice(),
["object-a".to_string()],
"the already returned page may be processed, but the task must not report success"
"the returned page is healed exactly once and the scan ends"
);
}
crates/heal/tests/heal_b5_versioned_regression_test.rs
+683
View File
@@ -0,0 +1,683 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! B5-2 (rustfs/backlog#919): real-disk-wipe e2e regression suite proving the
//! B5-1 version-aware heal enumeration physically repairs OLD non-latest
//! versions and DELETE-MARKER-latest objects.
//!
//! These drive the REAL `ECStoreHealStorage` (not a mock) against a real 4-disk
//! `ECStore`, mirroring `heal_integration_test.rs`. Every test is `#[serial]`
//! and re-runs the full `setup_test_env` init (which sets the process-global
//! bucket-metadata-sys OnceCell) — under `cargo nextest` each test runs
//! in its own process so the OnceCell never collides.
use http::HeaderMap;
use rustfs_common::heal_channel::{HealOpts, HealScanMode};
use rustfs_heal::heal::{
manager::{HealConfig, HealManager},
storage::{
ECStoreHealStorage, HealListItem, HealObjectOptions as ObjectOptions, HealPutObjReader as PutObjReader, HealStorageAPI,
},
task::{HealOptions, HealPriority, HealRequest, HealTaskStatus, HealType},
};
use serial_test::serial;
use std::{
path::{Path, PathBuf},
sync::{Arc, Once},
time::Duration,
};
use tokio::fs;
use tokio_util::sync::CancellationToken;
use tracing::info;
use walkdir::WalkDir;
mod storage_api;
use storage_api::integration::{
BucketOperations, BucketOptions, ECStore, Endpoint, EndpointServerPools, Endpoints, MakeBucketOptions, ObjectIO as _,
ObjectOperations as _, PoolEndpoints, init_bucket_metadata_sys, init_local_disks,
};
/// 256 KiB + change: large enough to be stored as non-inline erasure shards
/// (so each data version materializes as an on-disk `part.*` file we can assert
/// was physically restored on the wiped disk).
const NON_INLINE_TEST_DATA_SIZE: usize = 256 * 1024 + 137;
const SET_DISK_ID: &str = "pool_0_set_0";
fn versioned_test_data(seed: u8) -> Vec<u8> {
(0..NON_INLINE_TEST_DATA_SIZE)
.map(|idx| ((idx + seed as usize) % 251) as u8)
.collect()
}
static INIT: Once = Once::new();
fn init_tracing() {
INIT.call_once(|| {
let _ = tracing_subscriber::fmt()
.with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
.with_timer(tracing_subscriber::fmt::time::UtcTime::rfc_3339())
.with_thread_names(true)
.try_init();
});
}
/// Build a real 4-disk `ECStore` + `ECStoreHealStorage`. Mirrors
/// `heal_integration_test::setup_test_env`.
async fn setup_test_env() -> (Vec<PathBuf>, Arc<ECStore>, Arc<ECStoreHealStorage>) {
init_tracing();
let test_base_dir = format!("/tmp/rustfs_heal_b5_test_{}", uuid::Uuid::new_v4());
let temp_dir = PathBuf::from(&test_base_dir);
if temp_dir.exists() {
fs::remove_dir_all(&temp_dir).await.ok();
}
fs::create_dir_all(&temp_dir).await.unwrap();
let disk_paths = vec![
temp_dir.join("disk1"),
temp_dir.join("disk2"),
temp_dir.join("disk3"),
temp_dir.join("disk4"),
];
for disk_path in &disk_paths {
fs::create_dir_all(disk_path).await.unwrap();
}
let mut endpoints = Vec::new();
for (i, disk_path) in disk_paths.iter().enumerate() {
let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
endpoint.set_pool_index(0);
endpoint.set_set_index(0);
endpoint.set_disk_index(i);
endpoints.push(endpoint);
}
let pool_endpoints = PoolEndpoints {
legacy: false,
set_count: 1,
drives_per_set: 4,
endpoints: Endpoints::from(endpoints),
cmd_line: "test".to_string(),
platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
};
let endpoint_pools = EndpointServerPools::from(vec![pool_endpoints]);
init_local_disks(endpoint_pools.clone()).await.unwrap();
let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new())
.await
.unwrap();
let buckets_list = ecstore
.list_bucket(&BucketOptions {
no_metadata: true,
..Default::default()
})
.await
.unwrap();
let buckets = buckets_list.into_iter().map(|v| v.name).collect();
init_bucket_metadata_sys(ecstore.clone(), buckets).await;
let heal_storage = Arc::new(ECStoreHealStorage::new(ecstore.clone()));
(disk_paths, ecstore, heal_storage)
}
/// Create a bucket with S3 versioning ENABLED at creation time. Without this the
/// second PUT overwrites in place and DELETE removes the object outright — no
/// old versions and no delete-marker-latest would ever exist (empty-pass trap).
async fn create_versioned_bucket(ecstore: &Arc<ECStore>, bucket: &str) {
(**ecstore)
.make_bucket(
bucket,
&MakeBucketOptions {
versioning_enabled: true,
..Default::default()
},
)
.await
.expect("failed to create versioned bucket");
}
async fn create_unversioned_bucket(ecstore: &Arc<ECStore>, bucket: &str) {
(**ecstore)
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("failed to create unversioned bucket");
}
/// PUT a new version (versioned:true forces a fresh version id per write) and
/// return the created version id as a String.
async fn put_versioned(ecstore: &Arc<ECStore>, bucket: &str, object: &str, data: &[u8]) -> String {
let mut reader = PutObjReader::from_vec(data.to_vec());
let opts = ObjectOptions {
versioned: true,
..Default::default()
};
let info = (**ecstore)
.put_object(bucket, object, &mut reader, &opts)
.await
.expect("versioned put_object failed");
info.version_id
.map(|u| u.to_string())
.expect("versioned put must return a version id")
}
async fn put_unversioned(ecstore: &Arc<ECStore>, bucket: &str, object: &str, data: &[u8]) {
let mut reader = PutObjReader::from_vec(data.to_vec());
(**ecstore)
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
.await
.expect("unversioned put_object failed");
}
/// Create a delete-marker as the latest version (versioned:true, no version_id)
/// and return the delete-marker's version id.
async fn put_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> String {
let opts = ObjectOptions {
versioned: true,
..Default::default()
};
let info = (**ecstore)
.delete_object(bucket, object, opts)
.await
.expect("versioned delete (delete-marker) failed");
assert!(info.delete_marker, "delete on a versioned bucket must yield a delete marker");
info.version_id
.map(|u| u.to_string())
.expect("delete marker must carry a version id")
}
/// On-disk object directory for a given disk: `<disk>/<bucket>/<object>/`.
fn object_dir(disk: &Path, bucket: &str, object: &str) -> PathBuf {
disk.join(bucket).join(object)
}
/// Count `part.*` data-shard files two levels below the object dir
/// (`<object>/<data-uuid>/part.N`). One data dir per non-delete-marker version.
fn count_part_files(obj_dir: &Path) -> usize {
if !obj_dir.exists() {
return 0;
}
WalkDir::new(obj_dir)
.min_depth(2)
.max_depth(2)
.into_iter()
.filter_map(Result::ok)
.filter(|e| e.file_type().is_file() && e.file_name().to_str().map(|n| n.starts_with("part.")).unwrap_or(false))
.count()
}
fn xl_meta_path(obj_dir: &Path) -> PathBuf {
obj_dir.join("xl.meta")
}
fn recreate_heal_opts() -> HealOpts {
// Mirrors the proven-working object heal opts in
// `heal_integration_test::test_heal_format_with_data`.
HealOpts {
recreate: true,
remove: false,
..Default::default()
}
}
/// Enumerate every version via the REAL B5-1 paged listing, walking all pages.
async fn enumerate_all_versions(heal_storage: &Arc<ECStoreHealStorage>, bucket: &str) -> Vec<HealListItem> {
let mut items = Vec::new();
let mut token: Option<String> = None;
loop {
let (page, next, truncated) = heal_storage
.list_objects_for_heal_page(bucket, "", token.as_deref())
.await
.expect("list_objects_for_heal_page failed");
items.extend(page);
if !truncated {
break;
}
token = next;
if token.is_none() {
break;
}
}
items
}
/// Heal every enumerated version through the REAL `ECStoreHealStorage`, mirroring
/// the production per-version loop. Returns (healed_ok, failed).
async fn heal_all_versions(heal_storage: &Arc<ECStoreHealStorage>, bucket: &str) -> (usize, usize) {
let items = enumerate_all_versions(heal_storage, bucket).await;
let opts = recreate_heal_opts();
let mut healed = 0usize;
let mut failed = 0usize;
for item in items {
let (result, error) = heal_storage
.heal_object(bucket, &item.name, item.version_id.as_deref(), &opts)
.await
.expect("heal_object call itself must not error out");
if error.is_some() {
failed += 1;
info!(
"heal_object reported error for {}/{} v={:?} dm={}: {:?}",
bucket, item.name, item.version_id, item.is_delete_marker, error
);
} else {
healed += 1;
}
let _ = result;
}
(healed, failed)
}
async fn read_version(ecstore: &Arc<ECStore>, bucket: &str, object: &str, version_id: &str) -> Vec<u8> {
let opts = ObjectOptions {
version_id: Some(version_id.to_string()),
..Default::default()
};
let mut reader = ecstore
.get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
.await
.expect("failed to open version reader");
let mut buf = Vec::new();
tokio::io::copy(&mut reader, &mut buf)
.await
.expect("failed to read version data");
buf
}
mod serial_tests {
use super::*;
/// Directly exercises `ECStoreHealStorage::list_objects_for_heal_page` on a
/// real versioned fixture: two data versions + a delete-marker-latest. Proves
/// enumeration returns one `HealListItem` per version, flags the delete marker
/// via `is_delete_marker`, and carries the correct version ids.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[serial]
async fn test_enumeration_includes_all_versions_and_delete_marker_real_fixture() {
let (_disk_paths, ecstore, heal_storage) = setup_test_env().await;
let bucket = "b5-enum-versions";
let object = "obj.bin";
create_versioned_bucket(&ecstore, bucket).await;
let v1 = put_versioned(&ecstore, bucket, object, &versioned_test_data(1)).await;
let v2 = put_versioned(&ecstore, bucket, object, &versioned_test_data(2)).await;
let dm = put_delete_marker(&ecstore, bucket, object).await;
assert_ne!(v1, v2);
assert_ne!(v2, dm);
let items = enumerate_all_versions(&heal_storage, bucket).await;
// FIXTURE-NON-EMPTY GUARD: exactly the three versions we created.
assert_eq!(items.len(), 3, "must enumerate every version + the delete marker, got {items:?}");
assert!(items.iter().all(|it| it.name == object), "all entries belong to the same object");
let dm_items: Vec<&HealListItem> = items.iter().filter(|it| it.is_delete_marker).collect();
assert_eq!(dm_items.len(), 1, "exactly one entry must be flagged as a delete marker");
assert_eq!(
dm_items[0].version_id.as_deref(),
Some(dm.as_str()),
"the delete-marker entry must carry the delete marker's version id"
);
let ids: std::collections::HashSet<Option<String>> = items.iter().map(|it| it.version_id.clone()).collect();
assert!(ids.contains(&Some(v1.clone())), "old version v1 must be enumerated");
assert!(ids.contains(&Some(v2.clone())), "version v2 must be enumerated");
assert!(ids.contains(&Some(dm.clone())), "delete marker version must be enumerated");
// Every version id is a real (non-nil) id, so none normalized to None.
assert!(items.iter().all(|it| it.version_id.is_some()), "versioned entries must keep their ids");
}
/// Physical repair of an OLD non-latest version after wiping one disk.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[serial]
async fn test_heal_old_nonlatest_version_after_disk_wipe() {
let (disk_paths, ecstore, heal_storage) = setup_test_env().await;
let bucket = "b5-old-version-heal";
let object = "obj.bin";
create_versioned_bucket(&ecstore, bucket).await;
let data_v1 = versioned_test_data(10);
let data_v2 = versioned_test_data(20);
let v1 = put_versioned(&ecstore, bucket, object, &data_v1).await; // OLD, non-latest
let v2 = put_versioned(&ecstore, bucket, object, &data_v2).await; // latest
// ── Pre-wipe: prove the fixture actually has 2 versions on disk[0] ──
let obj_dir0 = object_dir(&disk_paths[0], bucket, object);
assert!(xl_meta_path(&obj_dir0).exists(), "xl.meta must exist before wipe");
assert_eq!(
count_part_files(&obj_dir0),
2,
"two data versions must each have an on-disk shard before wipe"
);
let pre_items = enumerate_all_versions(&heal_storage, bucket).await;
assert_eq!(pre_items.len(), 2, "fixture must expose both versions");
// ── Wipe disk[0]'s object dir (xl.meta + both data dirs) ──
std::fs::remove_dir_all(&obj_dir0).expect("failed to wipe object dir on disk[0]");
assert!(!obj_dir0.exists(), "object dir must be gone after wipe");
assert_eq!(count_part_files(&obj_dir0), 0, "no shards on disk[0] after wipe");
// ── Heal every enumerated version through the real heal storage ──
let (healed, failed) = heal_all_versions(&heal_storage, bucket).await;
assert_eq!(failed, 0, "no version may be recorded as failed");
assert!(healed >= 2, "both versions must be healed, healed={healed}");
// ── Post-heal: disk[0] physically restored, incl. the OLD version ──
assert!(xl_meta_path(&obj_dir0).exists(), "xl.meta must be restored on the wiped disk");
assert_eq!(
count_part_files(&obj_dir0),
2,
"both versions' data shards must be physically restored on disk[0]"
);
// Old non-latest version data must be intact and readable end-to-end.
assert_eq!(read_version(&ecstore, bucket, object, &v1).await, data_v1, "old version data corrupted");
assert_eq!(
read_version(&ecstore, bucket, object, &v2).await,
data_v2,
"latest version data corrupted"
);
}
/// Delete-marker-latest objects must be enumerated AND healed after a wipe.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[serial]
async fn test_heal_delete_marker_latest_enumerated_and_healed() {
let (disk_paths, ecstore, heal_storage) = setup_test_env().await;
let bucket = "b5-dm-latest-heal";
let object = "obj.bin";
create_versioned_bucket(&ecstore, bucket).await;
let data_v1 = versioned_test_data(30);
let v1 = put_versioned(&ecstore, bucket, object, &data_v1).await; // data version
let dm = put_delete_marker(&ecstore, bucket, object).await; // delete-marker latest
// ── Pre-wipe fixture proof: one data shard + a delete-marker latest ──
let obj_dir0 = object_dir(&disk_paths[0], bucket, object);
assert!(xl_meta_path(&obj_dir0).exists(), "xl.meta must exist before wipe");
assert_eq!(count_part_files(&obj_dir0), 1, "one data version => one shard before wipe");
let pre_items = enumerate_all_versions(&heal_storage, bucket).await;
assert_eq!(pre_items.len(), 2, "must enumerate the data version and the delete marker");
assert!(
pre_items
.iter()
.any(|it| it.is_delete_marker && it.version_id.as_deref() == Some(dm.as_str())),
"the delete-marker-latest must be enumerated as a heal unit"
);
// The latest, per get_object_info, is a delete marker (object reads as deleted).
let latest = ecstore.get_object_info(bucket, object, &ObjectOptions::default()).await;
assert!(
matches!(&latest, Ok(info) if info.delete_marker) || latest.is_err(),
"latest must resolve to a delete marker before heal"
);
// ── Wipe disk[0]'s object dir ──
std::fs::remove_dir_all(&obj_dir0).expect("failed to wipe object dir on disk[0]");
assert_eq!(count_part_files(&obj_dir0), 0, "no shards on disk[0] after wipe");
// ── Heal all enumerated versions (data version + delete marker) ──
let (healed, failed) = heal_all_versions(&heal_storage, bucket).await;
assert_eq!(failed, 0, "delete marker + data version must not be recorded as failed");
assert!(healed >= 2, "both the delete marker and the data version must heal, healed={healed}");
// ── Post-heal: xl.meta (with the tombstone) + the data shard restored ──
assert!(
xl_meta_path(&obj_dir0).exists(),
"xl.meta (carrying the delete-marker tombstone) must be restored on disk[0]"
);
assert_eq!(
count_part_files(&obj_dir0),
1,
"the underlying data version's shard must be physically restored on disk[0]"
);
// The old data version is still readable by id; the latest is still a DM.
assert_eq!(
read_version(&ecstore, bucket, object, &v1).await,
data_v1,
"healed data version corrupted"
);
let latest_after = ecstore.get_object_info(bucket, object, &ObjectOptions::default()).await;
assert!(
matches!(&latest_after, Ok(info) if info.delete_marker) || latest_after.is_err(),
"latest must remain a delete marker after heal"
);
}
/// Unversioned objects normalize to `version_id == None` and enumerate once
/// each (never a phantom `Some("null")`).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[serial]
async fn test_version_id_normalization_null_and_unversioned_real_fixture() {
let (_disk_paths, ecstore, heal_storage) = setup_test_env().await;
let bucket = "b5-unversioned-normalize";
create_unversioned_bucket(&ecstore, bucket).await;
put_unversioned(&ecstore, bucket, "a.bin", &versioned_test_data(1)).await;
put_unversioned(&ecstore, bucket, "b.bin", &versioned_test_data(2)).await;
// Overwrite b.bin: on an unversioned bucket this replaces in place, so
// there is still exactly one enumerable unit for it.
put_unversioned(&ecstore, bucket, "b.bin", &versioned_test_data(3)).await;
let items = enumerate_all_versions(&heal_storage, bucket).await;
assert_eq!(items.len(), 2, "unversioned bucket => exactly one heal unit per object, got {items:?}");
assert!(
items.iter().all(|it| it.version_id.is_none()),
"unversioned objects must normalize to version_id == None, never Some(\"null\"): {items:?}"
);
assert!(items.iter().all(|it| !it.is_delete_marker), "no delete markers expected");
let names: std::collections::HashSet<&str> = items.iter().map(|it| it.name.as_str()).collect();
assert!(names.contains("a.bin") && names.contains("b.bin"), "both objects enumerated once");
// NOTE: A literal MinIO-interop "null" *string* version id is only minted
// by the S3 request layer (versioning-suspended writes); it cannot be
// constructed through the internal ECStore put path used here (suspended/
// unversioned writes store no version id at all -> None). The nil-UUID ->
// None normalization itself is unit-covered in B5-1
// (crates/heal storage list_objects_for_heal_page maps
// `version_id.filter(|u| !u.is_nil())`). This test covers the real
// unversioned-object => None path end-to-end.
}
/// Unversioned bucket, full heal path via `HealManager` (recursive bucket
/// heal), after wiping one disk. Every object is healed exactly once and
/// remains readable; nothing is dropped or double-processed.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[serial]
async fn test_heal_unversioned_bucket_e2e() {
let (disk_paths, ecstore, heal_storage) = setup_test_env().await;
let bucket = "b5-unversioned-e2e";
create_unversioned_bucket(&ecstore, bucket).await;
let objects = ["o1.bin", "o2.bin", "o3.bin"];
let mut datas = Vec::new();
for (i, obj) in objects.iter().enumerate() {
let data = versioned_test_data(i as u8 + 40);
put_unversioned(&ecstore, bucket, obj, &data).await;
datas.push(data);
}
// Wipe every object dir on disk[0].
for obj in &objects {
let dir = object_dir(&disk_paths[0], bucket, obj);
std::fs::remove_dir_all(&dir).expect("failed to wipe object dir");
assert_eq!(count_part_files(&dir), 0);
}
// Enumeration returns exactly one unit per object (no duplicates).
let items = enumerate_all_versions(&heal_storage, bucket).await;
assert_eq!(items.len(), objects.len(), "one heal unit per object");
assert!(items.iter().all(|it| it.version_id.is_none()));
// Drive the real recursive bucket heal through the HealManager task loop.
let cfg = HealConfig {
heal_interval: Duration::from_millis(1),
..Default::default()
};
let heal_manager = HealManager::new(heal_storage.clone(), Some(cfg));
heal_manager.start().await.unwrap();
let request = HealRequest::new(
HealType::Bucket {
bucket: bucket.to_string(),
},
HealOptions {
recursive: true,
recreate_missing: true,
scan_mode: HealScanMode::Normal,
timeout: Some(Duration::from_secs(300)),
..Default::default()
},
HealPriority::Normal,
);
let task_id = request.id.clone();
let admission = heal_manager
.submit_heal_request(request)
.await
.expect("failed to submit unversioned bucket heal");
assert!(admission.is_admitted());
wait_for_task(&heal_manager, &task_id, Duration::from_secs(60)).await;
// Every object is physically restored on disk[0] and reads back intact.
for (obj, data) in objects.iter().zip(datas.iter()) {
let dir = object_dir(&disk_paths[0], bucket, obj);
assert!(xl_meta_path(&dir).exists(), "{obj}: xl.meta not restored on disk[0]");
assert_eq!(count_part_files(&dir), 1, "{obj}: data shard not restored on disk[0]");
let mut reader = ecstore
.get_object_reader(bucket, obj, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("healed object must be readable");
let mut buf = Vec::new();
tokio::io::copy(&mut reader, &mut buf).await.unwrap();
assert_eq!(&buf, data, "{obj}: healed data mismatch");
}
}
/// End-to-end drive of the REAL resume/checkpoint-backed erasure-set healer
/// (`ErasureSet` heal type -> `ErasureSetHealer::heal_bucket_with_resume`)
/// against real disks with a multi-version fixture across a wiped disk.
///
/// SCOPE NOTE: literally crossing the 1000-key `list_object_versions` page
/// boundary at e2e scale (1000+ real versions) is impractical for a unit-speed
/// test, so that exact boundary + mid-page cancel/resume behavior is covered
/// by the B5-1 in-crate loop tests
/// (`erasure_healer::resume_loop_tests::test_resume_across_page_boundary_no_drop_no_double`,
/// `test_object_with_versions_spanning_pages_advances`,
/// `test_non_advancing_cursor_aborts`,
/// `test_schedule_retry_resets_both_managers_and_reheals`). Here we prove the
/// same resume machinery drives a real ECStore heal to completion and repairs
/// every version, with the resume state cleaned up afterward.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[serial]
async fn test_heal_resume_across_page_boundary_e2e() {
let (disk_paths, ecstore, heal_storage) = setup_test_env().await;
let bucket = "b5-resume-e2e";
create_versioned_bucket(&ecstore, bucket).await;
// A small multi-object, multi-version fixture: 3 objects, 2 versions each,
// plus a delete-marker-latest on one of them.
let mut versions: Vec<(String, String, Vec<u8>)> = Vec::new(); // (object, version_id, data)
for obj_idx in 0..3u8 {
let object = format!("obj-{obj_idx}.bin");
let d1 = versioned_test_data(obj_idx + 50);
let d2 = versioned_test_data(obj_idx + 80);
let v1 = put_versioned(&ecstore, bucket, &object, &d1).await;
let v2 = put_versioned(&ecstore, bucket, &object, &d2).await;
versions.push((object.clone(), v1, d1));
versions.push((object.clone(), v2, d2));
}
// Delete-marker latest on obj-0.
let _dm = put_delete_marker(&ecstore, bucket, "obj-0.bin").await;
// Wipe disk[0]'s bucket dir entirely (all objects' shards + metadata).
let bucket_dir0 = disk_paths[0].join(bucket);
assert!(bucket_dir0.exists());
std::fs::remove_dir_all(&bucket_dir0).expect("failed to wipe bucket dir on disk[0]");
assert!(!bucket_dir0.exists());
// Drive the resume-backed erasure-set heal through the HealManager, which
// constructs the ErasureSetHealer and runs heal_bucket_with_resume.
let cfg = HealConfig {
heal_interval: Duration::from_millis(1),
..Default::default()
};
let heal_manager = HealManager::new(heal_storage.clone(), Some(cfg));
heal_manager.start().await.unwrap();
let request = HealRequest::new(
HealType::ErasureSet {
buckets: vec![bucket.to_string()],
set_disk_id: SET_DISK_ID.to_string(),
},
HealOptions {
recursive: true,
recreate_missing: true,
scan_mode: HealScanMode::Normal,
timeout: Some(Duration::from_secs(300)),
..Default::default()
},
HealPriority::Normal,
);
let task_id = request.id.clone();
let admission = heal_manager
.submit_heal_request(request)
.await
.expect("failed to submit erasure-set heal");
assert!(admission.is_admitted(), "erasure-set heal must be admitted");
wait_for_task(&heal_manager, &task_id, Duration::from_secs(120)).await;
// Every data version is readable end-to-end and physically restored on the
// wiped disk. (Resume machinery drove the full per-version heal.)
for (object, version_id, data) in &versions {
assert_eq!(
&read_version(&ecstore, bucket, object, version_id).await,
data,
"{object} v={version_id}: data not restored after resume-backed heal"
);
let dir = object_dir(&disk_paths[0], bucket, object);
assert!(xl_meta_path(&dir).exists(), "{object}: xl.meta not restored on disk[0]");
}
// Each object should have both of its data versions' shards back on disk[0].
for obj_idx in 0..3u8 {
let dir = object_dir(&disk_paths[0], bucket, &format!("obj-{obj_idx}.bin"));
assert_eq!(
count_part_files(&dir),
2,
"obj-{obj_idx}: both data versions' shards must be restored on disk[0]"
);
}
}
/// Poll a heal task to a terminal state, panicking on failure/timeout.
async fn wait_for_task(heal_manager: &HealManager, task_id: &str, timeout: Duration) {
let deadline = tokio::time::Instant::now() + timeout;
loop {
if let Ok(status) = heal_manager.get_task_status(task_id).await {
match status {
HealTaskStatus::Completed => return,
HealTaskStatus::Failed { ref error } => panic!("heal task failed: {error}"),
HealTaskStatus::Cancelled => panic!("heal task was cancelled"),
_ => {}
}
}
if tokio::time::Instant::now() >= deadline {
panic!("heal task {task_id} did not complete within {timeout:?}");
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
}
crates/heal/tests/heal_bug_fixes_test.rs
+6 -6
View File
@@ -162,7 +162,7 @@ fn test_path_to_str_helper() {
#[test]
fn test_heal_task_status_atomic_update() {
use rustfs_heal::heal::storage::{HealObjectInfo, HealStorageAPI};
use rustfs_heal::heal::storage::{HealListItem, HealObjectInfo, HealStorageAPI};
use rustfs_heal::heal::task::{HealOptions, HealRequest, HealTask, HealTaskStatus};
use std::sync::Arc;
@@ -234,7 +234,7 @@ fn test_heal_task_status_atomic_update() {
) -> rustfs_heal::Result<(rustfs_madmin::heal_commands::HealResultItem, Option<rustfs_heal::Error>)> {
Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None))
}
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result<Vec<String>> {
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result<Vec<HealListItem>> {
Ok(vec![])
}
async fn list_objects_for_heal_page(
@@ -242,7 +242,7 @@ fn test_heal_task_status_atomic_update() {
_bucket: &str,
_prefix: &str,
_continuation_token: Option<&str>,
) -> rustfs_heal::Result<(Vec<String>, Option<String>, bool)> {
) -> rustfs_heal::Result<(Vec<HealListItem>, Option<String>, bool)> {
Ok((vec![], None, false))
}
async fn get_disk_for_resume(&self, _set_disk_id: &str) -> rustfs_heal::Result<DiskStore> {
@@ -277,7 +277,7 @@ fn test_heal_task_status_atomic_update() {
#[tokio::test]
async fn test_heal_task_transient_object_exists_skip_avoids_recreate() {
use rustfs_heal::heal::storage::{DiskStatus, HealObjectInfo, HealStorageAPI};
use rustfs_heal::heal::storage::{DiskStatus, HealListItem, HealObjectInfo, HealStorageAPI};
use rustfs_heal::heal::task::{HealOptions, HealPriority, HealRequest, HealTask, HealTaskStatus, HealType};
use std::sync::{
Arc,
@@ -376,7 +376,7 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() {
Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None))
}
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result<Vec<String>> {
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result<Vec<HealListItem>> {
Ok(Vec::new())
}
@@ -385,7 +385,7 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() {
_bucket: &str,
_prefix: &str,
_continuation_token: Option<&str>,
) -> rustfs_heal::Result<(Vec<String>, Option<String>, bool)> {
) -> rustfs_heal::Result<(Vec<HealListItem>, Option<String>, bool)> {
Ok((Vec::new(), None, false))
}
crates/heal/tests/storage_api.rs
+1
View File
@@ -26,6 +26,7 @@ pub(crate) mod integration {
pub(crate) use rustfs_ecstore::api::storage::init_local_disks;
pub(crate) use rustfs_storage_api::BucketOperations;
pub(crate) use rustfs_storage_api::BucketOptions;
pub(crate) use rustfs_storage_api::MakeBucketOptions;
pub(crate) use rustfs_storage_api::ObjectIO;
pub(crate) use rustfs_storage_api::ObjectOperations;
}