Compare commits

..

1 Commits

Author SHA1 Message Date
马登山 23b85b792e fix(scanner): unify unknown metadata size accounting 2026-08-22 22:58:05 +08:00
24 changed files with 1284 additions and 1923 deletions
+89 -1
View File
@@ -307,6 +307,38 @@ pub struct DiskUsageStatus {
pub snapshot_exists: bool,
}
/// A bounded reconciliation record for an object whose logical size could not
/// be trusted at the scanner boundary. The scanner persists these records in
/// its cache; keeping the model here avoids a second, incompatible accounting
/// representation in storage-facing crates.
#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct SizeReconciliationEntry {
/// Stable object/version identity key (not a metrics label).
pub key: String,
pub bucket: String,
pub object: String,
#[serde(default)]
pub version_id: Option<String>,
#[serde(default)]
pub generation: Option<String>,
/// Structured reason label; raw metadata values must never be stored here.
pub reason: String,
#[serde(default)]
pub physical_size: Option<u64>,
#[serde(default)]
pub first_seen: u64,
#[serde(default)]
pub attempts: u32,
}
/// Object scope refreshed by one scanner pass. Existing debts in this scope
/// are removed before the pass's unresolved records are inserted.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct SizeReconciliationScope {
pub bucket: String,
pub object: String,
}
/// Size summary for a single object or group of objects
#[derive(Debug, Default, Clone)]
pub struct SizeSummary {
@@ -336,6 +368,16 @@ pub struct SizeSummary {
pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
/// Per-tier accounting, keyed by storage class or remote tier name
pub tier_stats: HashMap<String, TierStats>,
/// Size-resolution debts observed while scanning this summary.
pub size_reconciliation: Vec<SizeReconciliationEntry>,
/// True when the per-object summary exceeded its bounded debt buffer.
/// Callers must retain prior ledger entries rather than treating the
/// partial list as a complete refresh.
pub size_reconciliation_truncated: bool,
/// Object scopes refreshed by this summary. They let the durable ledger
/// remove versions that resolved without allocating one key per healthy
/// version on the hot path.
pub reconciliation_scopes: Vec<SizeReconciliationScope>,
}
/// Replication target size summary
@@ -830,7 +872,8 @@ impl DataUsageEntry {
///
/// The canonical wire format is written by the hand-written map-encoded
/// `Serialize` on the scanner-side `DataUsageCacheInfo`
/// (`crates/scanner/src/data_usage_define.rs`), which carries 16 fields.
/// (`crates/scanner/src/data_usage_define.rs`), which carries the original 16
/// fields plus an optional reconciliation field.
/// This type decodes only the shared subset and is deliberately not
/// `Serialize`: a derived (array) encoding of this 6-field subset would
/// corrupt the cache for scanner readers, so no write path may exist here.
@@ -1774,6 +1817,51 @@ impl SizeSummary {
entry.pending_count = entry.pending_count.saturating_add(stats.pending_count);
entry.failed_count = entry.failed_count.saturating_add(stats.failed_count);
}
for entry in &other.size_reconciliation {
self.record_size_reconciliation(entry.clone());
}
self.size_reconciliation_truncated |= other.size_reconciliation_truncated;
for scope in &other.reconciliation_scopes {
self.record_reconciliation_scope(&scope.bucket, &scope.object);
}
}
/// Add one reconciliation debt, coalescing repeated observations in the
/// same object summary. The scanner cache applies its own larger bound.
pub fn record_size_reconciliation(&mut self, entry: SizeReconciliationEntry) {
const MAX_SUMMARY_RECONCILIATION_ENTRIES: usize = 1024;
if let Some(existing) = self.size_reconciliation.iter_mut().find(|value| value.key == entry.key) {
existing.reason = entry.reason;
existing.physical_size = entry.physical_size;
existing.generation = entry.generation;
existing.version_id = entry.version_id;
return;
}
if self.size_reconciliation.len() < MAX_SUMMARY_RECONCILIATION_ENTRIES {
self.size_reconciliation.push(entry);
} else {
self.size_reconciliation_truncated = true;
}
}
/// Mark one object scope as refreshed. Duplicate scopes are suppressed so
/// merging summaries remains bounded and deterministic.
pub fn record_reconciliation_scope(&mut self, bucket: &str, object: &str) {
if !self
.reconciliation_scopes
.iter()
.any(|scope| scope.bucket == bucket && scope.object == object)
{
if self.reconciliation_scopes.len() >= 1024 {
self.size_reconciliation_truncated = true;
return;
}
self.reconciliation_scopes.push(SizeReconciliationScope {
bucket: bucket.to_string(),
object: object.to_string(),
});
}
}
}
@@ -3425,44 +3425,6 @@ mod tests {
assert!(mutexes.contains_key("second"));
}
#[tokio::test]
async fn update_all_targets_publishes_disable_proxy_on_target_client() {
// The read-proxy selector (replication_proxy::get_proxy_targets) skips
// targets whose TargetClient carries disable_proxy — the persisted
// per-target opt-out must survive client publication.
let sys = BucketTargetSys::default();
let target = |arn: &str, disable_proxy: bool| BucketTarget {
arn: arn.to_string(),
endpoint: "192.168.1.10:9000".to_string(),
target_bucket: "target-bucket".to_string(),
region: "us-east-1".to_string(),
disable_proxy,
credentials: Some(Credentials {
access_key: "access".to_string(),
secret_key: "secret".to_string(),
session_token: None,
expiration: None,
}),
..Default::default()
};
let targets = BucketTargets {
targets: vec![target("arn:proxied", false), target("arn:opted-out", true)],
};
sys.update_all_targets("bucket", Some(&targets)).await;
let proxied = sys
.get_remote_target_client("bucket", "arn:proxied")
.await
.expect("client should be published");
assert!(!proxied.disable_proxy);
let opted_out = sys
.get_remote_target_client("bucket", "arn:opted-out")
.await
.expect("client should be published");
assert!(opted_out.disable_proxy, "disable_proxy must reach the published TargetClient");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn target_updates_serialize_client_build_through_publication_per_bucket() {
let sys = Arc::new(BucketTargetSys::default());
+2 -276
View File
@@ -297,16 +297,10 @@ impl ECStore {
#[cfg(test)]
mod tests {
use super::*;
use crate::bucket::metadata_sys;
use crate::core::pools::{PoolDecommissionInfo, PoolStatus};
use crate::disk::{DeleteOptions, DiskOption, format::FormatV3, new_disk};
use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
use crate::runtime::instance::InstanceContext;
use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations};
use crate::disk::{DiskOption, format::FormatV3, new_disk};
use crate::layout::endpoints::{Endpoints, PoolEndpoints};
use crate::store::init_format::{load_format_erasure, save_format_file};
use crate::store::init_local_disks_with_instance_ctx;
use tokio_util::sync::CancellationToken;
async fn minimal_heal_pool(pool_idx: usize) -> Arc<Sets> {
let format = FormatV3::new(1, 1);
@@ -353,51 +347,6 @@ mod tests {
}
}
async fn multi_pool_heal_store() -> (tempfile::TempDir, Arc<ECStore>, CancellationToken) {
let temp_dir = tempfile::tempdir().expect("multi-pool heal test directory should be created");
let mut pool_endpoints = Vec::new();
for pool_index in 0..2 {
let mut endpoints = Vec::new();
for disk_index in 0..4 {
let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}"));
tokio::fs::create_dir_all(&disk_path)
.await
.expect("multi-pool heal test disk should be created");
let mut endpoint = Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8"))
.expect("test endpoint should parse");
endpoint.set_pool_index(pool_index);
endpoint.set_set_index(0);
endpoint.set_disk_index(disk_index);
endpoints.push(endpoint);
}
pool_endpoints.push(PoolEndpoints {
legacy: false,
set_count: 1,
drives_per_set: 4,
endpoints: Endpoints::from(endpoints),
cmd_line: format!("heal-owner-pool-{pool_index}"),
platform: "test".to_string(),
});
}
let endpoint_pools = EndpointServerPools::from(pool_endpoints);
let instance_ctx = Arc::new(InstanceContext::new());
init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
.await
.expect("multi-pool local disks should initialize");
let shutdown = CancellationToken::new();
let store = ECStore::new_with_instance_ctx(
"127.0.0.1:0".parse().expect("test address should parse"),
endpoint_pools,
shutdown.clone(),
instance_ctx,
)
.await
.expect("multi-pool test store should initialize");
metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
(temp_dir, store, shutdown)
}
#[tokio::test]
async fn heal_object_pool_scope_selects_only_requested_pool() {
let store = minimal_heal_store().await;
@@ -557,229 +506,6 @@ mod tests {
}
}
#[tokio::test]
#[serial_test::serial]
async fn unscoped_heal_object_suspended_owner_semantics() {
let (_temp_dir, store, shutdown) = multi_pool_heal_store().await;
let bucket = format!("heal-owner-{}", Uuid::new_v4().simple());
let active_object = "active-owner";
let suspended_only_object = "suspended-only";
let duplicate_object = "duplicate-owner";
let marker_object = "marker-owner";
let quorum_object = "quorum-owner";
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("bucket should be created in all pools");
let mut active_reader = PutObjReader::from_vec(b"active owner".to_vec());
store.pools[0]
.put_object(&bucket, active_object, &mut active_reader, &ObjectOptions::default())
.await
.expect("active owner object should be written");
let active_disks = store.pools[0].disk_set[0].disks.read().await.clone();
let missing_active_disk = active_disks[0].clone().expect("active disk should be online");
missing_active_disk
.delete(
&bucket,
active_object,
DeleteOptions {
recursive: true,
immediate: true,
..Default::default()
},
)
.await
.expect("active owner shard should be removed for repair");
assert!(
missing_active_disk.read_xl(&bucket, active_object, false).await.is_err(),
"the active owner fixture must start with one missing metadata copy"
);
let mut suspended_reader = PutObjReader::from_vec(b"suspended owner".to_vec());
store.pools[1]
.put_object(&bucket, suspended_only_object, &mut suspended_reader, &ObjectOptions::default())
.await
.expect("suspended owner object should be written");
for (pool_index, mod_time) in [1_i64, 2_i64].into_iter().enumerate() {
let mut duplicate_reader = PutObjReader::from_vec(format!("duplicate-pool-{pool_index}").into_bytes());
store.pools[pool_index]
.put_object(
&bucket,
duplicate_object,
&mut duplicate_reader,
&ObjectOptions {
mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(mod_time)),
..Default::default()
},
)
.await
.expect("duplicate owner object should be written");
}
let duplicate_missing_disk = store.pools[0].disk_set[0].disks.read().await[0]
.clone()
.expect("duplicate active owner disk should be online");
duplicate_missing_disk
.delete(
&bucket,
duplicate_object,
DeleteOptions {
recursive: true,
immediate: true,
..Default::default()
},
)
.await
.expect("duplicate active owner shard should be removed for repair");
let history_version = Uuid::new_v4();
let mut history_reader = PutObjReader::from_vec(b"marker history".to_vec());
store.pools[0]
.put_object(
&bucket,
marker_object,
&mut history_reader,
&ObjectOptions {
versioned: true,
version_id: Some(history_version.to_string()),
mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1)),
..Default::default()
},
)
.await
.expect("versioned marker history should be written");
store.pools[0]
.delete_object(
&bucket,
marker_object,
ObjectOptions {
versioned: true,
mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(2)),
..Default::default()
},
)
.await
.expect("delete marker should be written");
let mut quorum_reader = PutObjReader::from_vec(b"quorum boundary".to_vec());
store.pools[0]
.put_object(&bucket, quorum_object, &mut quorum_reader, &ObjectOptions::default())
.await
.expect("quorum boundary object should be written");
{
let mut pool_meta = store.pool_meta.write().await;
let mut next = PoolMeta::new(&store.pools, &pool_meta);
next.pools[1].decommission = Some(PoolDecommissionInfo {
start_time: Some(OffsetDateTime::UNIX_EPOCH),
..Default::default()
});
*pool_meta = next;
}
let (_, duplicate_owner) = store
.get_latest_object_info_with_idx(&bucket, duplicate_object, &ObjectOptions::default())
.await
.expect("duplicate owner should resolve");
assert_eq!(duplicate_owner, 1, "latest duplicate must win when all pools are eligible");
let (_, active_duplicate_owner) = store
.get_latest_object_info_with_idx(
&bucket,
duplicate_object,
&ObjectOptions {
skip_decommissioned: true,
..Default::default()
},
)
.await
.expect("active duplicate owner should resolve");
assert_eq!(
active_duplicate_owner, 0,
"suspended duplicate must be excluded from active owner selection"
);
let (duplicate_result, duplicate_err) = store
.handle_heal_object(&bucket, duplicate_object, "", &HealOpts::default())
.await
.expect("duplicate owner heal should complete through the production path");
assert_eq!(duplicate_result.object, duplicate_object);
assert!(duplicate_err.is_none(), "active duplicate should be repaired: {duplicate_err:?}");
assert!(
duplicate_missing_disk.read_xl(&bucket, duplicate_object, false).await.is_ok(),
"production heal must repair the active duplicate owner rather than the suspended owner"
);
let (marker_info, marker_owner) = store
.get_latest_object_info_with_idx(
&bucket,
marker_object,
&ObjectOptions {
skip_decommissioned: true,
versioned: true,
..Default::default()
},
)
.await
.expect("latest delete marker should resolve");
assert_eq!(marker_owner, 0);
assert!(marker_info.delete_marker, "latest version must preserve delete-marker semantics");
let (active_result, active_err) = store
.handle_heal_object(&bucket, active_object, "", &HealOpts::default())
.await
.expect("unscoped active-owner heal should complete");
assert_eq!(active_result.object, active_object);
assert!(active_err.is_none(), "active owner must be selected even with a suspended pool");
assert!(
missing_active_disk.read_xl(&bucket, active_object, false).await.is_ok(),
"active owner heal must write the missing disk metadata: result={active_result:?}, err={active_err:?}"
);
assert!(
store.pools[1]
.get_object_info(&bucket, active_object, &ObjectOptions::default())
.await
.is_err(),
"the suspended pool must not be written for an active-owner object"
);
let (suspended_result, suspended_err) = store
.handle_heal_object(&bucket, suspended_only_object, "", &HealOpts::default())
.await
.expect("unscoped suspended-only heal should return a terminal result");
assert!(suspended_result.object.is_empty());
assert!(matches!(suspended_err, Some(Error::FileNotFound)));
assert!(
store.pools[1]
.get_object_info(&bucket, suspended_only_object, &ObjectOptions::default())
.await
.is_ok(),
"suspended-only data must remain untouched when unscoped heal reports absent"
);
let (_, explicit_err) = store
.handle_heal_object(
&bucket,
suspended_only_object,
"",
&HealOpts {
pool: Some(1),
..Default::default()
},
)
.await
.expect("explicit suspended-owner heal should return a mapped error");
assert!(matches!(explicit_err, Some(Error::SlowDown)));
let original_quorum_disks = store.pools[0].disk_set[0].disks.read().await.clone();
let surviving_quorum_disk = original_quorum_disks[3].clone();
*store.pools[0].disk_set[0].disks.write().await = vec![None, None, None, surviving_quorum_disk];
let (_, quorum_err) = store
.handle_heal_object(&bucket, quorum_object, "", &HealOpts::default())
.await
.expect("quorum boundary heal should return a mapped result");
*store.pools[0].disk_set[0].disks.write().await = original_quorum_disks;
assert!(
matches!(quorum_err, Some(Error::ErasureReadQuorum)),
"quorum-boundary heal must preserve quorum error, got {quorum_err:?}"
);
shutdown.cancel();
}
#[tokio::test]
async fn handle_heal_format_continues_after_a_pool_error() {
let canonical_format = FormatV3::new(1, 3);
+53 -509
View File
@@ -13,9 +13,9 @@
// limitations under the License.
use crate::heal::{
progress::{HealProgress, add_bytes, increment_counter},
progress::HealProgress,
resume::{
CheckpointManager, CheckpointObjectOutcome, ReplacementTargetIdentity, ResumeManager, ResumeUtils, compose_key,
CheckpointManager, ReplacementTargetIdentity, ResumeManager, ResumeUtils, compose_key,
replacement_target_identities_match,
},
storage::{HealStorageAPI, next_heal_listing_token},
@@ -410,9 +410,6 @@ impl ErasureSetHealer {
&& state.successful_objects == 0
&& state.failed_objects == 0
&& state.skipped_objects == 0
&& state.skipped_new_versions == 0
&& state.skipped_ilm_expired == 0
&& state.processed_bytes == 0
{
// schedule_retry persists the authoritative resume reset before
// resetting the checkpoint. Reapply the checkpoint reset after
@@ -477,23 +474,6 @@ impl ErasureSetHealer {
// 2. initialize progress
self.initialize_progress(buckets, &state).await;
let (baseline_known, baseline_count, baseline_size, baseline_generation) = {
let baseline = self.progress.read().await;
(
baseline.baseline_known,
baseline.objects_total_count,
baseline.objects_total_size,
baseline.baseline_generation,
)
};
if baseline_known {
resume_manager
.set_progress_baseline(baseline_count, baseline_size, baseline_generation)
.await?;
checkpoint_manager
.set_progress_baseline(baseline_count, baseline_size, baseline_generation)
.await?;
}
// 3. continue from checkpoint
let current_bucket_index = checkpoint.current_bucket_index;
@@ -503,66 +483,12 @@ impl ErasureSetHealer {
let mut successful_objects = state.successful_objects;
let mut failed_objects = state.failed_objects;
let mut skipped_objects = state.skipped_objects;
let checkpoint_has_progress = checkpoint.baseline_known
|| checkpoint.successful_objects > 0
|| checkpoint.failed_object_count > 0
|| checkpoint.skipped_object_count > 0
|| checkpoint.skipped_new_versions > 0
|| checkpoint.skipped_ilm_expired > 0
|| checkpoint.processed_bytes > 0
|| checkpoint.total_objects > 0
|| checkpoint.total_bytes > 0
|| checkpoint.baseline_generation.is_some()
|| checkpoint.counter_unknown;
let checkpoint_generation_mismatch = checkpoint.baseline_known && checkpoint.baseline_generation != baseline_generation;
let mut restored_counter_unknown = state.counter_unknown || checkpoint.counter_unknown;
if checkpoint_has_progress {
successful_objects = checkpoint.successful_objects;
failed_objects = checkpoint.failed_object_count;
skipped_objects = checkpoint.skipped_object_count;
let restored_processed_objects = successful_objects
.checked_add(failed_objects)
.and_then(|value| value.checked_add(skipped_objects))
.and_then(|value| value.checked_add(checkpoint.skipped_new_versions))
.and_then(|value| value.checked_add(checkpoint.skipped_ilm_expired));
let checkpoint_counter_overflow = restored_processed_objects.is_none();
restored_counter_unknown |= checkpoint_counter_overflow;
processed_objects = restored_processed_objects.unwrap_or(u64::MAX);
let mut progress = self.progress.write().await;
progress.objects_scanned = processed_objects;
progress.objects_healed = successful_objects;
progress.objects_failed = failed_objects;
progress.skipped_objects = skipped_objects;
progress.skipped_new_versions = checkpoint.skipped_new_versions;
progress.skipped_ilm_expired = checkpoint.skipped_ilm_expired;
if checkpoint.baseline_known && !checkpoint_generation_mismatch {
progress.objects_total_count = checkpoint.total_objects;
progress.objects_total_size = checkpoint.total_bytes;
progress.baseline_generation = checkpoint.baseline_generation;
progress.baseline_known = true;
}
progress.bytes_processed = checkpoint.processed_bytes;
progress.counter_unknown = state.counter_unknown || checkpoint.counter_unknown;
progress.refresh_progress_percentage();
if checkpoint_generation_mismatch || checkpoint_counter_overflow || progress.counter_unknown {
progress.mark_unknown();
}
}
if checkpoint_generation_mismatch {
restored_counter_unknown = true;
}
if restored_counter_unknown {
checkpoint_manager.mark_counter_unknown().await?;
resume_manager.mark_counter_unknown().await?;
}
let mut failed_buckets = 0u64;
// 4. process remaining buckets
for (bucket_idx, bucket) in buckets.iter().enumerate().skip(current_bucket_index) {
// check if completed
if state.completed_buckets.contains(bucket) {
checkpoint_manager.complete_bucket(bucket_idx.saturating_add(1)).await?;
current_object_index = 0;
continue;
}
@@ -590,42 +516,13 @@ impl ErasureSetHealer {
return bucket_result;
}
// update progress
let progress_snapshot = self.progress.read().await;
let bytes_processed = progress_snapshot.bytes_processed;
let skipped_new_versions = progress_snapshot.skipped_new_versions;
let skipped_ilm_expired = progress_snapshot.skipped_ilm_expired;
let counter_unknown = progress_snapshot.counter_unknown;
drop(progress_snapshot);
// The checkpoint is the recovery authority for object progress.
// Publish its counters and fence before the resume summary so a
// crash between the two stores cannot make recovery select newer
// summary bytes with an older checkpoint ledger.
if counter_unknown {
checkpoint_manager.mark_counter_unknown().await?;
}
checkpoint_manager
.update_progress(successful_objects, failed_objects, skipped_objects, bytes_processed)
.await?;
checkpoint_manager
.set_skipped_version_counts(skipped_new_versions, skipped_ilm_expired)
.await?;
// update checkpoint position
checkpoint_manager.update_position(bucket_idx, current_object_index).await?;
// update progress
resume_manager
.update_progress_with_bytes(
processed_objects,
successful_objects,
failed_objects,
skipped_objects,
bytes_processed,
)
.update_progress(processed_objects, successful_objects, failed_objects, skipped_objects)
.await?;
resume_manager
.set_skipped_version_counts(skipped_new_versions, skipped_ilm_expired)
.await?;
if counter_unknown {
resume_manager.mark_counter_unknown().await?;
}
// check cancel status
if self.cancel_token.is_cancelled() {
@@ -645,7 +542,6 @@ impl ErasureSetHealer {
match bucket_result {
Ok(_) => {
resume_manager.complete_bucket(bucket).await?;
checkpoint_manager.complete_bucket(bucket_idx.saturating_add(1)).await?;
debug!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
@@ -671,9 +567,7 @@ impl ErasureSetHealer {
error = %e,
"Erasure set bucket heal failed"
);
// A single durable cursor and ledger cannot safely preserve
// this bucket while processing a later one.
break;
// continue to next bucket, do not interrupt the whole process
}
}
@@ -881,48 +775,20 @@ impl ErasureSetHealer {
// 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.failed_objects.contains(&key)
|| checkpoint.skipped_objects.contains(&key)
{
if checkpoint.processed_objects.contains(&key) || checkpoint.skipped_objects.contains(&key) {
continue;
}
if should_skip_new_version(item.mod_time_unix_nanos, started_at_secs) {
let counter_ok = increment_counter(processed_objects);
checkpoint_manager.add_processed_object(key).await?;
*processed_objects = processed_objects.saturating_add(1);
completed_in_page = completed_in_page.saturating_add(1);
counter!("rustfs_heal_skipped_new_versions_total").increment(1);
let (skipped_new, skipped_ilm, counter_unknown) = {
{
let mut progress = self.progress.write().await;
progress.record_skipped_new_version();
progress.set_current_object(Some(format!("skipped_new: {bucket}/{}", item.name)));
progress.update_object_progress(
*processed_objects,
*successful_objects,
*failed_objects,
*skipped_objects,
bytes_processed,
);
if !counter_ok {
progress.mark_unknown();
}
(progress.skipped_new_versions, progress.skipped_ilm_expired, progress.counter_unknown)
};
checkpoint_manager
.record_object_outcome(
key,
CheckpointObjectOutcome::Processed,
*successful_objects,
*failed_objects,
*skipped_objects,
bytes_processed,
skipped_new,
skipped_ilm,
!counter_ok || counter_unknown,
)
.await?;
if !counter_ok || counter_unknown {
resume_manager.mark_counter_unknown().await?;
progress.update_progress(*processed_objects, *successful_objects, *failed_objects, bytes_processed);
}
debug!(
target: "rustfs::heal::erasure_healer",
@@ -954,40 +820,15 @@ impl ErasureSetHealer {
)
.await?
{
let counter_ok = increment_counter(processed_objects);
checkpoint_manager.add_processed_object(key).await?;
*processed_objects = processed_objects.saturating_add(1);
completed_in_page = completed_in_page.saturating_add(1);
counter!("rustfs_heal_skipped_ilm_expired_total").increment(1);
let (skipped_new, skipped_ilm, counter_unknown) = {
{
let mut progress = self.progress.write().await;
progress.record_skipped_ilm_expired();
progress.set_current_object(Some(format!("skipped_ilm: {bucket}/{}", item.name)));
progress.update_object_progress(
*processed_objects,
*successful_objects,
*failed_objects,
*skipped_objects,
bytes_processed,
);
if !counter_ok {
progress.mark_unknown();
}
(progress.skipped_new_versions, progress.skipped_ilm_expired, progress.counter_unknown)
};
checkpoint_manager
.record_object_outcome(
key,
CheckpointObjectOutcome::Processed,
*successful_objects,
*failed_objects,
*skipped_objects,
bytes_processed,
skipped_new,
skipped_ilm,
!counter_ok || counter_unknown,
)
.await?;
if !counter_ok || counter_unknown {
resume_manager.mark_counter_unknown().await?;
progress.update_progress(*processed_objects, *successful_objects, *failed_objects, bytes_processed);
}
debug!(
target: "rustfs::heal::erasure_healer",
@@ -1113,11 +954,11 @@ impl ErasureSetHealer {
while let Some((key, object, version_id, result)) = page_tasks.next().await {
let (object_size, result) = result;
let mut telemetry_unknown = false;
let checkpoint_outcome = match result {
match result {
Ok(true) => {
telemetry_unknown |= !increment_counter(successful_objects);
telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size);
*successful_objects += 1;
bytes_processed = bytes_processed.saturating_add(object_size);
checkpoint_manager.add_processed_object(key).await?;
debug!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
@@ -1130,11 +971,11 @@ impl ErasureSetHealer {
state = "healed",
"Erasure set object healed"
);
CheckpointObjectOutcome::Processed
}
Ok(false) => {
telemetry_unknown |= !increment_counter(successful_objects);
telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size);
checkpoint_manager.add_processed_object(key).await?;
*successful_objects += 1;
bytes_processed = bytes_processed.saturating_add(object_size);
debug!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
@@ -1147,12 +988,12 @@ impl ErasureSetHealer {
state = "missing_treated_as_ok",
"Erasure set missing object treated as ok"
);
CheckpointObjectOutcome::Processed
}
Err(err @ Error::TaskCancelled) | Err(err @ Error::TaskTimeout) => return Err(err),
Err(Error::TransientSkip { message }) => {
telemetry_unknown |= !increment_counter(skipped_objects);
telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size);
*skipped_objects += 1;
bytes_processed = bytes_processed.saturating_add(object_size);
checkpoint_manager.add_skipped_object(key).await?;
demote_to_debug_when!(!take_failure_log_sample(&mut transient_skip_samples_logged), warn, target: "rustfs::heal::erasure_healer", {
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
@@ -1165,11 +1006,11 @@ impl ErasureSetHealer {
error = %message,
"Erasure set object heal skipped due to transient error"
});
CheckpointObjectOutcome::Skipped
}
Err(err) => {
telemetry_unknown |= !increment_counter(failed_objects);
telemetry_unknown |= !add_bytes(&mut bytes_processed, object_size);
*failed_objects += 1;
bytes_processed = bytes_processed.saturating_add(object_size);
checkpoint_manager.add_failed_object(key).await?;
demote_to_debug_when!(!take_failure_log_sample(&mut failure_samples_logged), warn, target: "rustfs::heal::erasure_healer", {
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
@@ -1182,42 +1023,15 @@ impl ErasureSetHealer {
error = %err,
"Erasure set object heal failed"
});
CheckpointObjectOutcome::Failed
}
};
}
telemetry_unknown |= !increment_counter(processed_objects);
*processed_objects += 1;
completed_in_page += 1;
let (progress_unknown, skipped_new_versions, skipped_ilm_expired) = {
{
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("{bucket}/{object}")));
progress.update_object_progress(
*processed_objects,
*successful_objects,
*failed_objects,
*skipped_objects,
bytes_processed,
);
if telemetry_unknown {
progress.mark_unknown();
}
(progress.counter_unknown, progress.skipped_new_versions, progress.skipped_ilm_expired)
};
checkpoint_manager
.record_object_outcome(
key,
checkpoint_outcome,
*successful_objects,
*failed_objects,
*skipped_objects,
bytes_processed,
skipped_new_versions,
skipped_ilm_expired,
telemetry_unknown || progress_unknown,
)
.await?;
if telemetry_unknown || progress_unknown {
resume_manager.mark_counter_unknown().await?;
progress.update_progress(*processed_objects, *successful_objects, *failed_objects, bytes_processed);
}
if completed_in_page.is_multiple_of(100) {
@@ -1227,22 +1041,16 @@ impl ErasureSetHealer {
*current_object_index = global_obj_idx;
// Persist the checkpoint ledger and page position before exposing
// the next resume cursor. A crash before cursor publication keeps
// the page identities available for exact-once replay.
checkpoint_manager.advance_page(bucket_index, *current_object_index).await?;
// 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?;
// Check if there are more pages
if !is_truncated {
break;
}
continuation_token = next_heal_listing_token(bucket, "", next_token, is_truncated)?;
if continuation_token.is_none() {
// A truncated page without a continuation token is terminal.
// Retain its ledger until bucket completion is durable.
break;
}
resume_manager.set_resume_cursor(continuation_token.clone()).await?;
checkpoint_manager.prune_completed_page().await?;
// Anti-loop guard: an empty page reported as truncated cannot advance
// the cursor (there is no last identity to move past), so treat it as a
@@ -1261,6 +1069,12 @@ impl ErasureSetHealer {
)));
}
previous_page_last = page_last;
continuation_token = next_heal_listing_token(bucket, "", next_token, is_truncated)?;
if continuation_token.is_none() {
// Truncated but no continuation token: treat as end of listing.
break;
}
}
Ok(())
@@ -1269,66 +1083,10 @@ impl ErasureSetHealer {
/// initialize progress tracking
async fn initialize_progress(&self, _buckets: &[String], state: &crate::heal::resume::ResumeState) {
let mut progress = self.progress.write().await;
let existing_baseline = (
progress.objects_total_count,
progress.objects_total_size,
progress.baseline_generation,
progress.progress_state,
progress.baseline_known,
);
let baseline_generation_mismatch =
state.baseline_known && existing_baseline.4 && state.baseline_generation != existing_baseline.2;
let use_persisted_baseline = state.baseline_known && !baseline_generation_mismatch;
progress.objects_scanned = state.processed_objects;
progress.objects_scanned = state.total_objects;
progress.objects_healed = state.successful_objects;
progress.objects_failed = state.failed_objects;
progress.skipped_objects = state.skipped_objects;
progress.skipped_new_versions = state.skipped_new_versions;
progress.skipped_ilm_expired = state.skipped_ilm_expired;
progress.bytes_processed = state.processed_bytes;
progress.counter_unknown = state.counter_unknown;
if use_persisted_baseline
|| existing_baseline.0 > 0
|| existing_baseline.1 > 0
|| existing_baseline.2.is_some()
|| existing_baseline.4
{
progress.objects_total_count = if use_persisted_baseline {
state.total_objects
} else {
existing_baseline.0
};
progress.objects_total_size = if use_persisted_baseline {
state.total_bytes
} else {
existing_baseline.1
};
progress.baseline_generation = if use_persisted_baseline {
state.baseline_generation
} else {
existing_baseline.2
};
progress.baseline_known = use_persisted_baseline
|| existing_baseline.0 > 0
|| existing_baseline.1 > 0
|| existing_baseline.2.is_some()
|| existing_baseline.4;
}
progress.progress_state = if use_persisted_baseline
|| existing_baseline.0 > 0
|| existing_baseline.1 > 0
|| existing_baseline.2.is_some()
|| existing_baseline.4
{
crate::heal::progress::HealProgressState::Running
} else {
crate::heal::progress::HealProgressState::Indeterminate
};
if baseline_generation_mismatch || state.counter_unknown {
progress.mark_unknown();
}
progress.ledger_complete = false;
progress.refresh_progress_percentage();
progress.bytes_processed = 0; // Resume state tracks object counts, not byte counters.
progress.start_time = UNIX_EPOCH.checked_add(Duration::from_secs(state.start_time));
progress.last_update_time = UNIX_EPOCH.checked_add(Duration::from_secs(state.last_update));
progress.set_current_object(state.current_object.clone());
@@ -1506,8 +1264,8 @@ mod resume_loop_tests {
};
use crate::heal::progress::HealProgress;
use crate::heal::resume::{
CheckpointManager, CheckpointObjectOutcome, RESUME_CHECKPOINT_FILE, ReplacementTargetIdentity, ResumeDeleteFailure,
ResumeManager, ResumeUtils, compose_key,
CheckpointManager, RESUME_CHECKPOINT_FILE, ReplacementTargetIdentity, ResumeDeleteFailure, ResumeManager, ResumeUtils,
compose_key,
};
use crate::heal::storage::{HealLifecycleExpiryContext, HealListItem, HealObjectInfo, HealStorageAPI};
use crate::heal::storage_api::status::BucketInfo;
@@ -1647,7 +1405,6 @@ mod resume_loop_tests {
list_include_lifecycle_object_info: Mutex<Vec<bool>>,
replacement_target_identity_sequences: Mutex<VecDeque<Vec<ReplacementTargetIdentity>>>,
fail_listing: AtomicBool,
fail_listing_buckets: Mutex<HashSet<String>>,
}
impl FakeStorage {
@@ -1684,9 +1441,6 @@ mod resume_loop_tests {
fn fail_listing(&self) {
self.fail_listing.store(true, Ordering::SeqCst);
}
fn fail_bucket_listing(&self, bucket: &str) {
self.fail_listing_buckets.lock().unwrap().insert(bucket.to_string());
}
}
#[async_trait::async_trait]
@@ -1777,7 +1531,7 @@ mod resume_loop_tests {
}
async fn list_objects_for_heal_page(
&self,
bucket: &str,
_bucket: &str,
_prefix: &str,
continuation_token: Option<&str>,
include_lifecycle_object_info: bool,
@@ -1786,7 +1540,7 @@ mod resume_loop_tests {
.lock()
.unwrap()
.push(include_lifecycle_object_info);
if self.fail_listing.load(Ordering::SeqCst) || self.fail_listing_buckets.lock().unwrap().contains(bucket) {
if self.fail_listing.load(Ordering::SeqCst) {
return Err(Error::other("injected listing failure"));
}
let key = continuation_token.map(str::to_string);
@@ -2164,49 +1918,6 @@ mod resume_loop_tests {
assert!(state.completed_buckets.is_empty(), "the failed bucket must remain resumable");
}
#[tokio::test]
async fn bucket_failure_stops_before_a_later_bucket_checkpoint() {
let env = make_env().await;
let task_id = ResumeUtils::generate_task_id();
let buckets = vec!["a".to_string(), "b".to_string()];
let resume = ResumeManager::new(
env.healer.disk.clone(),
task_id.clone(),
"erasure_set".to_string(),
"pool_0_set_0".to_string(),
buckets.clone(),
)
.await
.unwrap();
let checkpoint = CheckpointManager::new(env.healer.disk.clone(), task_id.clone())
.await
.unwrap();
env.storage.fail_bucket_listing("a");
for _ in 0..3 {
assert!(resume.schedule_retry().await.unwrap());
}
env.healer
.execute_heal_with_resume(&buckets, "pool_0_set_0", &resume, &checkpoint)
.await
.expect_err("the first bucket failure must keep the pass incomplete");
let persisted = checkpoint.get_checkpoint().await;
assert_eq!(persisted.current_bucket_index, 0);
assert!(resume.get_state().await.completed_buckets.is_empty());
let resumed = ResumeManager::load_from_disk(env.healer.disk.clone(), &task_id)
.await
.unwrap();
let checkpoint = CheckpointManager::load_from_disk(env.healer.disk.clone(), &task_id)
.await
.unwrap();
env.healer
.execute_heal_with_resume(&buckets, "pool_0_set_0", &resumed, &checkpoint)
.await
.expect_err("recovery must retry the earlier failed bucket");
assert!(!resumed.get_state().await.completed);
}
#[tokio::test]
async fn completed_resume_state_is_not_selected_for_a_new_heal() {
let env = make_env().await;
@@ -2396,175 +2107,8 @@ mod resume_loop_tests {
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");
// Keep the final page cursor until the outer loop durably completes the
// bucket, so a crash can replay only this page against its identities.
assert_eq!(env.resume.resume_cursor().await, Some("t1".to_string()));
}
#[tokio::test]
async fn persisted_failure_waits_for_the_bounded_retry_after_page_replay() {
let env = make_env().await;
env.storage.set_page(
None,
Page {
items: vec![item("object", Some("v1"), false)],
next: None,
truncated: false,
},
);
env.checkpoint
.record_object_outcome(
compose_key("object", Some("v1")),
CheckpointObjectOutcome::Failed,
0,
1,
0,
0,
0,
0,
false,
)
.await
.unwrap();
env.checkpoint.advance_page(0, 1).await.unwrap();
let resumed = ResumeManager::load_from_disk(env.healer.disk.clone(), &env.task_id)
.await
.unwrap();
let checkpoint = CheckpointManager::load_from_disk(env.healer.disk.clone(), &env.task_id)
.await
.unwrap();
env.healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &resumed, &checkpoint)
.await
.expect_err("the persisted failure must schedule a bounded retry");
assert!(
env.storage.calls().is_empty(),
"the failed identity must not be repeated in the same pass"
);
env.healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &resumed, &checkpoint)
.await
.expect("the bounded retry must heal the object");
assert_eq!(env.storage.calls(), vec![("object".to_string(), Some("v1".to_string()))]);
let state = resumed.get_state().await;
assert_eq!(state.successful_objects, 1);
assert_eq!(state.failed_objects, 0);
}
#[tokio::test]
async fn final_page_crash_replays_only_the_retained_page_identities() {
let env = make_env().await;
env.storage.set_page(
None,
Page {
items: vec![item("first", Some("v1"), false)],
next: Some("final-page".to_string()),
truncated: true,
},
);
env.storage.set_page(
Some("final-page"),
Page {
items: vec![item("last", Some("v1"), false)],
next: None,
truncated: false,
},
);
let (processed, successful, failed, skipped, result) = run(&env).await;
result.expect("the bucket pass must finish before the simulated crash");
assert_eq!((processed, successful, failed, skipped), (2, 2, 0, 0));
let resumed = ResumeManager::load_from_disk(env.healer.disk.clone(), &env.task_id)
.await
.unwrap();
let checkpoint = CheckpointManager::load_from_disk(env.healer.disk.clone(), &env.task_id)
.await
.unwrap();
env.healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &resumed, &checkpoint)
.await
.expect("the retained final-page ledger must make recovery exact");
assert_eq!(
env.storage.calls(),
vec![
("first".to_string(), Some("v1".to_string())),
("last".to_string(), Some("v1".to_string()))
]
);
let state = resumed.get_state().await;
assert_eq!(state.successful_objects, 2);
assert_eq!(state.processed_objects, 2);
}
#[tokio::test]
async fn truncated_page_without_token_retains_its_replay_ledger() {
let env = make_env().await;
env.storage.set_page(
None,
Page {
items: vec![item("object", Some("v1"), false)],
next: None,
truncated: true,
},
);
let (processed, successful, failed, skipped, result) = run(&env).await;
result.expect("the tokenless truncated page is a terminal page");
assert_eq!((processed, successful, failed, skipped), (1, 1, 0, 0));
let resumed = ResumeManager::load_from_disk(env.healer.disk.clone(), &env.task_id)
.await
.unwrap();
let checkpoint = CheckpointManager::load_from_disk(env.healer.disk.clone(), &env.task_id)
.await
.unwrap();
env.healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &resumed, &checkpoint)
.await
.expect("terminal-page recovery must not replay a durable identity");
assert_eq!(env.storage.calls(), vec![("object".to_string(), Some("v1".to_string()))]);
let state = resumed.get_state().await;
assert_eq!(state.successful_objects, 1);
assert_eq!(state.processed_objects, 1);
}
#[tokio::test]
async fn completed_bucket_reconciles_its_final_page_checkpoint_after_crash() {
let env = make_env().await;
env.storage.set_page(
None,
Page {
items: vec![item("object", Some("v1"), false)],
next: None,
truncated: false,
},
);
let (_, _, _, _, result) = run(&env).await;
result.expect("the bucket pass must finish before the simulated crash");
env.resume.complete_bucket("b").await.unwrap();
let resumed = ResumeManager::load_from_disk(env.healer.disk.clone(), &env.task_id)
.await
.unwrap();
let checkpoint = CheckpointManager::load_from_disk(env.healer.disk.clone(), &env.task_id)
.await
.unwrap();
env.healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &resumed, &checkpoint)
.await
.expect("recovery must finish the checkpoint transition without replaying the bucket");
assert_eq!(env.storage.calls(), vec![("object".to_string(), Some("v1".to_string()))]);
let checkpoint = checkpoint.get_checkpoint().await;
assert_eq!(checkpoint.current_bucket_index, 1);
assert!(checkpoint.processed_objects.is_empty());
// Final page not truncated => cursor cleared.
assert_eq!(env.resume.resume_cursor().await, None);
}
#[tokio::test]
+9 -66
View File
@@ -2012,44 +2012,16 @@ impl HealManager {
}
let mut snapshot = HealProgress::default();
let mut has_object_sweep = false;
let mut all_object_baselines_known = true;
let mut counter_overflow = false;
let mut stage_current = 0_u64;
let mut stage_total = 0_u64;
for task in active_tasks {
let progress = task.get_progress().await;
let object_sweep = matches!(progress.kind, crate::heal::progress::HealProgressKind::ObjectSweep);
has_object_sweep |= object_sweep;
if object_sweep {
all_object_baselines_known &= progress.baseline_known;
}
counter_overflow |=
progress.counter_unknown || matches!(progress.progress_state, crate::heal::progress::HealProgressState::Unknown);
match stage_current.checked_add(progress.stage_current) {
Some(sum) => stage_current = sum,
None => counter_overflow = true,
}
match stage_total.checked_add(progress.stage_total) {
Some(sum) => stage_total = sum,
None => counter_overflow = true,
}
for (target, value) in [
(&mut snapshot.objects_scanned, progress.objects_scanned),
(&mut snapshot.objects_healed, progress.objects_healed),
(&mut snapshot.objects_failed, progress.objects_failed),
(&mut snapshot.skipped_objects, progress.skipped_objects),
(&mut snapshot.skipped_new_versions, progress.skipped_new_versions),
(&mut snapshot.skipped_ilm_expired, progress.skipped_ilm_expired),
(&mut snapshot.objects_total_count, progress.objects_total_count),
(&mut snapshot.objects_total_size, progress.objects_total_size),
(&mut snapshot.bytes_processed, progress.bytes_processed),
] {
match target.checked_add(value) {
Some(sum) => *target = sum,
None => counter_overflow = true,
}
}
snapshot.objects_scanned = snapshot.objects_scanned.saturating_add(progress.objects_scanned);
snapshot.objects_healed = snapshot.objects_healed.saturating_add(progress.objects_healed);
snapshot.objects_failed = snapshot.objects_failed.saturating_add(progress.objects_failed);
snapshot.skipped_new_versions = snapshot.skipped_new_versions.saturating_add(progress.skipped_new_versions);
snapshot.skipped_ilm_expired = snapshot.skipped_ilm_expired.saturating_add(progress.skipped_ilm_expired);
snapshot.objects_total_count = snapshot.objects_total_count.saturating_add(progress.objects_total_count);
snapshot.objects_total_size = snapshot.objects_total_size.saturating_add(progress.objects_total_size);
snapshot.bytes_processed = snapshot.bytes_processed.saturating_add(progress.bytes_processed);
snapshot.start_time = match (snapshot.start_time, progress.start_time) {
(Some(current), Some(next)) => Some(current.min(next)),
(None, next) => next,
@@ -2064,36 +2036,7 @@ impl HealManager {
snapshot.current_object = progress.current_object;
}
}
snapshot.kind = if has_object_sweep {
crate::heal::progress::HealProgressKind::ObjectSweep
} else {
crate::heal::progress::HealProgressKind::Stage
};
snapshot.stage_current = stage_current;
snapshot.stage_total = stage_total;
snapshot.baseline_known = has_object_sweep && all_object_baselines_known;
snapshot.progress_state = if counter_overflow {
crate::heal::progress::HealProgressState::Unknown
} else if has_object_sweep && !all_object_baselines_known {
crate::heal::progress::HealProgressState::Indeterminate
} else if has_object_sweep {
crate::heal::progress::HealProgressState::Running
} else if stage_total == 0 {
crate::heal::progress::HealProgressState::Indeterminate
} else {
crate::heal::progress::HealProgressState::Running
};
if counter_overflow {
snapshot.progress_percentage = 0.0;
} else if !has_object_sweep {
snapshot.progress_percentage = if stage_total == 0 {
0.0
} else {
((stage_current as f64 / stage_total as f64) * 100.0).min(99.999)
};
} else {
snapshot.refresh_progress_percentage();
}
snapshot.refresh_progress_percentage();
snapshot.refresh_estimated_completion_time();
Some(snapshot)
}
+33 -316
View File
@@ -15,70 +15,15 @@
use serde::{Deserialize, Serialize};
use std::time::{Duration, SystemTime};
pub(crate) fn increment_counter(counter: &mut u64) -> bool {
match counter.checked_add(1) {
Some(next) => {
*counter = next;
true
}
None => {
*counter = u64::MAX;
false
}
}
}
pub(crate) fn add_bytes(total: &mut u64, amount: u64) -> bool {
match total.checked_add(amount) {
Some(next) => {
*total = next;
true
}
None => {
*total = u64::MAX;
false
}
}
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum HealProgressKind {
#[default]
Unknown,
Stage,
ObjectSweep,
}
/// Whether the object ledger can produce a meaningful percentage.
///
/// A zero-valued baseline is not a completed scan: it means that no complete
/// usage snapshot was available. Keep this state explicit so callers do not
/// mistake the legacy `0.0` wire value for a measured zero-percent result.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum HealProgressState {
#[default]
Unknown,
Indeterminate,
Running,
Completed,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct HealProgress {
#[serde(default)]
pub kind: HealProgressKind,
/// Objects scanned
pub objects_scanned: u64,
/// Objects healed
pub objects_healed: u64,
/// Objects failed
pub objects_failed: u64,
/// Versions deferred for a later retry pass.
#[serde(default)]
pub skipped_objects: u64,
/// Versions skipped because they were written after this heal started
pub skipped_new_versions: u64,
/// Versions skipped because lifecycle already selected them for expiry
@@ -99,38 +44,11 @@ pub struct HealProgress {
pub last_update_time: Option<SystemTime>,
/// Estimated completion time
pub estimated_completion_time: Option<SystemTime>,
/// Current stage number. Stage updates are intentionally independent from
/// the object ledger below.
#[serde(default)]
pub stage_current: u64,
/// Number of stages in the current task.
#[serde(default)]
pub stage_total: u64,
/// Explicitly distinguishes a missing usage baseline from measured 0%.
#[serde(default)]
pub progress_state: HealProgressState,
/// True only after the task's durable completion ledger was committed.
#[serde(default)]
pub ledger_complete: bool,
/// Generation of the usage snapshot used for the baseline, if available.
#[serde(default)]
pub baseline_generation: Option<u64>,
/// Whether the baseline was explicitly observed. This is separate from
/// the counters so a known empty scope (0 objects, 0 bytes) is not
/// confused with a legacy snapshot that omitted the baseline fields.
#[serde(default)]
pub baseline_known: bool,
/// Internal telemetry fence set when an aggregate counter overflows or
/// becomes inconsistent. It prevents a later refresh from fabricating a
/// percentage from the poisoned values.
#[serde(default)]
pub counter_unknown: bool,
}
impl HealProgress {
pub fn new() -> Self {
Self {
kind: HealProgressKind::Unknown,
start_time: Some(SystemTime::now()),
last_update_time: Some(SystemTime::now()),
..Default::default()
@@ -138,87 +56,12 @@ impl HealProgress {
}
pub fn update_progress(&mut self, scanned: u64, healed: u64, failed: u64, bytes: u64) {
self.update_object_sweep_progress(scanned, healed, failed, bytes);
}
pub fn update_object_sweep_progress(&mut self, scanned: u64, healed: u64, failed: u64, bytes: u64) {
self.kind = HealProgressKind::ObjectSweep;
self.objects_scanned = scanned;
self.objects_healed = healed;
self.objects_failed = failed;
self.bytes_processed = bytes;
self.last_update_time = Some(SystemTime::now());
let explicit_skipped = match self.skipped_new_versions.checked_add(self.skipped_ilm_expired) {
Some(value) => value,
None => {
self.mark_unknown();
0
}
};
let skipped = healed
.checked_add(failed)
.and_then(|value| value.checked_add(explicit_skipped))
.and_then(|value| scanned.checked_sub(value))
.unwrap_or(0);
self.update_object_progress(scanned, healed, failed, skipped, bytes);
}
/// Update task stage progress without modifying object counters.
pub fn update_stage(&mut self, current: u64, total: u64) {
let object_sweep_active = matches!(self.kind, HealProgressKind::ObjectSweep);
if !object_sweep_active {
self.kind = HealProgressKind::Stage;
}
self.ledger_complete = false;
self.stage_current = current.min(total);
self.stage_total = total;
if object_sweep_active {
self.last_update_time = Some(SystemTime::now());
self.refresh_progress_percentage();
return;
}
self.progress_state = if total == 0 {
HealProgressState::Indeterminate
} else {
HealProgressState::Running
};
self.progress_percentage = if total == 0 {
0.0
} else {
(current as f64 / total as f64 * 100.0).min(100.0)
};
self.last_update_time = Some(SystemTime::now());
}
/// Update the disjoint object ledger. `scanned` is the number of terminal
/// object outcomes and must equal healed + failed + deferred skipped plus
/// the two terminal skip classes. Overflow is a corrupt/unknown counter
/// state, not a reason to abort a completed heal.
pub fn update_object_progress(&mut self, scanned: u64, healed: u64, failed: u64, skipped: u64, bytes: u64) {
self.kind = HealProgressKind::ObjectSweep;
// `skipped` is the transient/deferred class. The two explicit skip
// counters are terminal classifications too, so include them in the
// same ledger without making callers maintain a second aggregate.
let outcomes = healed
.checked_add(failed)
.and_then(|value| value.checked_add(skipped))
.and_then(|value| value.checked_add(self.skipped_new_versions))
.and_then(|value| value.checked_add(self.skipped_ilm_expired));
self.objects_scanned = scanned;
self.objects_healed = healed;
self.objects_failed = failed;
self.skipped_objects = skipped;
self.bytes_processed = bytes;
self.last_update_time = Some(SystemTime::now());
self.ledger_complete = false;
if outcomes != Some(scanned) {
// Telemetry corruption must not abort a heal. Preserve the
// counters for diagnostics, but do not derive a percentage from a
// double-counted or overflowing ledger.
self.mark_unknown();
return;
}
self.refresh_progress_percentage();
self.refresh_estimated_completion_time();
}
@@ -226,88 +69,50 @@ impl HealProgress {
pub fn set_total_baseline(&mut self, objects_total_count: u64, objects_total_size: u64) {
self.objects_total_count = objects_total_count;
self.objects_total_size = objects_total_size;
self.baseline_known = true;
self.last_update_time = Some(SystemTime::now());
self.refresh_progress_percentage();
self.refresh_estimated_completion_time();
}
pub fn set_total_baseline_with_generation(&mut self, objects_total_count: u64, objects_total_size: u64, generation: u64) {
self.baseline_generation = Some(generation);
self.set_total_baseline(objects_total_count, objects_total_size);
}
pub fn record_skipped_new_version(&mut self) {
let Some(next) = self.skipped_new_versions.checked_add(1) else {
self.mark_unknown();
return;
};
self.skipped_new_versions = next;
self.skipped_new_versions = self.skipped_new_versions.saturating_add(1);
self.last_update_time = Some(SystemTime::now());
self.refresh_progress_percentage();
self.refresh_estimated_completion_time();
}
pub fn record_skipped_ilm_expired(&mut self) {
let Some(next) = self.skipped_ilm_expired.checked_add(1) else {
self.mark_unknown();
return;
};
self.skipped_ilm_expired = next;
self.skipped_ilm_expired = self.skipped_ilm_expired.saturating_add(1);
self.last_update_time = Some(SystemTime::now());
self.refresh_progress_percentage();
self.refresh_estimated_completion_time();
}
fn completed_for_baseline(&self) -> Option<u64> {
fn completed_for_baseline(&self) -> u64 {
self.objects_healed
.checked_add(self.objects_failed)?
.checked_add(self.skipped_objects)?
.checked_add(self.skipped_new_versions)?
.checked_add(self.skipped_ilm_expired)
.saturating_add(self.objects_failed)
.saturating_add(self.skipped_new_versions)
.saturating_add(self.skipped_ilm_expired)
}
pub(crate) fn refresh_progress_percentage(&mut self) {
if self.ledger_complete {
self.progress_state = HealProgressState::Completed;
self.progress_percentage = 100.0;
return;
}
if self.counter_unknown {
self.progress_state = HealProgressState::Unknown;
self.progress_percentage = 0.0;
return;
}
if !self.baseline_known {
self.progress_state = HealProgressState::Indeterminate;
self.progress_percentage = 0.0;
self.estimated_completion_time = None;
return;
}
if self.objects_total_size > 0 {
self.progress_percentage = ((self.bytes_processed as f64 / self.objects_total_size as f64) * 100.0).min(100.0);
self.progress_percentage = self.progress_percentage.min(99.999);
self.progress_state = HealProgressState::Running;
return;
}
if self.objects_total_count > 0 {
let Some(completed) = self.completed_for_baseline() else {
self.progress_state = HealProgressState::Unknown;
self.progress_percentage = 0.0;
return;
};
let completed = self.completed_for_baseline();
self.progress_percentage = ((completed as f64 / self.objects_total_count as f64) * 100.0).min(100.0);
self.progress_percentage = self.progress_percentage.min(99.999);
self.progress_state = HealProgressState::Running;
return;
}
if self.baseline_known {
self.progress_state = HealProgressState::Running;
self.progress_percentage = 0.0;
return;
let total = self
.objects_scanned
.saturating_add(self.objects_healed)
.saturating_add(self.objects_failed);
if total > 0 {
self.progress_percentage = (self.objects_healed as f64 / total as f64) * 100.0;
}
self.progress_state = HealProgressState::Indeterminate;
self.progress_percentage = 0.0;
}
pub fn set_current_object(&mut self, object: Option<String>) {
@@ -320,11 +125,7 @@ impl HealProgress {
self.estimated_completion_time = None;
return;
};
if self.is_completed()
|| self.progress_percentage <= 0.0
|| self.progress_percentage >= 100.0
|| self.bytes_processed == 0
{
if self.is_completed() || !(0.0..100.0).contains(&self.progress_percentage) || self.bytes_processed == 0 {
self.estimated_completion_time = None;
return;
}
@@ -341,39 +142,18 @@ impl HealProgress {
}
pub fn is_completed(&self) -> bool {
self.ledger_complete
}
/// Mark telemetry unknown while allowing the underlying heal operation to
/// continue. This is used for corrupt/overflowing counters at the
/// observability boundary; it must never turn a successful heal into an
/// execution error.
pub fn mark_unknown(&mut self) {
self.counter_unknown = true;
self.progress_state = HealProgressState::Unknown;
self.ledger_complete = false;
self.progress_percentage = 0.0;
self.estimated_completion_time = None;
self.last_update_time = Some(SystemTime::now());
}
/// Mark the object ledger terminal only after the enclosing task has
/// committed all durable resume state and cleanup fences.
pub fn mark_completed(&mut self) {
let telemetry_unknown = self.counter_unknown || self.progress_state == HealProgressState::Unknown;
self.ledger_complete = true;
if !telemetry_unknown {
self.progress_state = HealProgressState::Completed;
if self.progress_percentage >= 100.0 {
return true;
}
self.progress_percentage = 100.0;
self.last_update_time = Some(SystemTime::now());
self.estimated_completion_time = None;
if self.objects_total_count > 0 || self.objects_total_size > 0 {
return false;
}
self.objects_scanned > 0 && self.objects_healed.saturating_add(self.objects_failed) >= self.objects_scanned
}
pub fn get_success_rate(&self) -> f64 {
let Some(total) = self.objects_healed.checked_add(self.objects_failed) else {
return 0.0;
};
let total = self.objects_healed + self.objects_failed;
if total > 0 {
(self.objects_healed as f64 / total as f64) * 100.0
} else {
@@ -450,7 +230,6 @@ mod tests {
assert_eq!(progress.objects_scanned, 0);
assert_eq!(progress.objects_healed, 0);
assert_eq!(progress.objects_failed, 0);
assert_eq!(progress.skipped_objects, 0);
assert_eq!(progress.skipped_new_versions, 0);
assert_eq!(progress.skipped_ilm_expired, 0);
assert_eq!(progress.objects_total_count, 0);
@@ -471,8 +250,10 @@ mod tests {
assert_eq!(progress.objects_healed, 8);
assert_eq!(progress.objects_failed, 2);
assert_eq!(progress.bytes_processed, 1024);
assert_eq!(progress.progress_state, HealProgressState::Indeterminate);
assert_eq!(progress.progress_percentage, 0.0);
// Progress percentage should be calculated based on healed/total
// total = scanned + healed + failed = 10 + 8 + 2 = 20
// healed/total = 8/20 = 0.4 = 40%
assert!((progress.progress_percentage - 40.0).abs() < 0.001);
assert!(progress.last_update_time.is_some());
}
@@ -481,8 +262,7 @@ mod tests {
let mut progress = HealProgress::new();
progress.start_time = Some(SystemTime::now() - Duration::from_secs(10));
progress.set_total_baseline(100, 16384);
progress.update_progress(25, 25, 0, 4096);
progress.update_progress(100, 25, 0, 4096);
let eta = progress
.estimated_completion_time
@@ -495,7 +275,7 @@ mod tests {
let mut progress = HealProgress::new();
progress.set_total_baseline(10, 8192);
progress.update_progress(25, 25, 0, 4096);
progress.update_progress(100, 25, 0, 4096);
assert!((progress.progress_percentage - 50.0).abs() < 0.001);
}
@@ -505,7 +285,7 @@ mod tests {
let mut progress = HealProgress::new();
progress.set_total_baseline(10, 0);
progress.update_progress(5, 3, 2, 0);
progress.update_progress(100, 3, 2, 0);
assert!((progress.progress_percentage - 50.0).abs() < 0.001);
}
@@ -515,7 +295,7 @@ mod tests {
let mut progress = HealProgress::new();
progress.set_total_baseline(10, 0);
progress.update_progress(5, 3, 2, 0);
progress.update_progress(100, 3, 2, 0);
progress.record_skipped_new_version();
assert_eq!(progress.skipped_new_versions, 1);
@@ -556,8 +336,7 @@ mod tests {
fn test_heal_progress_update_progress_all_healed() {
let mut progress = HealProgress::new();
// When scanned=0, healed=10, failed=0: total=10, progress = 10/10 = 100%
progress.update_progress(10, 10, 0, 2048);
progress.mark_completed();
progress.update_progress(0, 10, 0, 2048);
// All healed, should be 100%
assert!((progress.progress_percentage - 100.0).abs() < 0.001);
@@ -615,7 +394,6 @@ mod tests {
assert_eq!(json["objectsScanned"], 10);
assert_eq!(json["objectsHealed"], 8);
assert_eq!(json["objectsFailed"], 2);
assert_eq!(json["skippedObjects"], 0);
assert_eq!(json["skippedNewVersions"], 0);
assert_eq!(json["skippedIlmExpired"], 0);
assert_eq!(json["bytesProcessed"], 1024);
@@ -627,7 +405,6 @@ mod tests {
fn test_heal_progress_is_completed_by_percentage() {
let mut progress = HealProgress::new();
progress.update_progress(10, 10, 0, 1024);
progress.mark_completed();
assert!(progress.is_completed());
}
@@ -638,7 +415,7 @@ mod tests {
progress.objects_scanned = 10;
progress.objects_healed = 8;
progress.objects_failed = 2;
progress.mark_completed();
// healed + failed = 8 + 2 = 10 >= scanned = 10
assert!(progress.is_completed());
}
@@ -678,66 +455,6 @@ mod tests {
assert!((progress.get_success_rate() - 100.0).abs() < 0.001);
}
#[test]
fn single_object_progress_reaches_terminal_100() {
let mut progress = HealProgress::new();
progress.update_object_progress(1, 1, 0, 0, 128);
assert!(!progress.is_completed());
progress.mark_completed();
assert!(progress.is_completed());
assert_eq!(progress.progress_percentage, 100.0);
}
#[test]
fn progress_without_baseline_is_indeterminate() {
let mut progress = HealProgress::new();
progress.update_object_progress(1, 1, 0, 0, 128);
assert_eq!(progress.progress_state, HealProgressState::Indeterminate);
assert_eq!(progress.progress_percentage, 0.0);
assert!(progress.estimated_completion_time.is_none());
}
#[test]
fn progress_retry_is_exactly_once() {
let mut progress = HealProgress::new();
progress.set_total_baseline(1, 128);
progress.update_object_progress(1, 1, 0, 0, 128);
progress.update_object_progress(1, 1, 0, 0, 128);
assert_eq!(progress.objects_scanned, 1);
assert_eq!(progress.objects_healed, 1);
assert_eq!(progress.bytes_processed, 128);
}
#[test]
fn progress_never_triggers_cleanup_before_terminal_ledger_empty() {
let mut progress = HealProgress::new();
progress.progress_percentage = 100.0;
assert!(!progress.is_completed());
progress.mark_completed();
assert!(progress.is_completed());
}
#[test]
fn progress_counter_overflow_is_marked_unknown_without_aborting_completed_heal() {
let mut progress = HealProgress::new();
progress.update_object_progress(u64::MAX, u64::MAX, 1, 0, 0);
assert_eq!(progress.progress_state, HealProgressState::Unknown);
progress.mark_completed();
assert!(progress.is_completed());
assert_eq!(progress.progress_state, HealProgressState::Unknown);
}
#[test]
fn stage_updates_do_not_double_count_object_outcomes() {
let mut progress = HealProgress::new();
progress.update_object_progress(2, 1, 0, 1, 256);
progress.update_stage(3, 4);
assert_eq!(progress.kind, HealProgressKind::ObjectSweep);
assert_eq!(progress.objects_scanned, 2);
assert_eq!(progress.objects_healed, 1);
assert_eq!(progress.skipped_objects, 1);
}
#[test]
fn test_heal_statistics_new() {
let stats = HealStatistics::new();
+4 -130
View File
@@ -31,7 +31,7 @@ mod checkpoint;
mod replacement;
mod utils;
pub use checkpoint::{CheckpointManager, CheckpointObjectOutcome, ResumeCheckpoint};
pub use checkpoint::{CheckpointManager, ResumeCheckpoint};
pub(crate) use replacement::replacement_target_identities_match;
use replacement::replacement_targets_match_identities;
pub use replacement::{
@@ -340,12 +340,6 @@ pub struct ResumeState {
pub failed_objects: u64,
/// skipped objects
pub skipped_objects: u64,
/// Terminal versions skipped because they were newer than the heal start.
#[serde(default)]
pub skipped_new_versions: u64,
/// Terminal versions handed to lifecycle expiry.
#[serde(default)]
pub skipped_ilm_expired: u64,
/// current bucket
pub current_bucket: Option<String>,
/// current object
@@ -360,24 +354,6 @@ pub struct ResumeState {
pub retry_count: u32,
/// max retries
pub max_retries: u32,
/// Bytes accounted by the object ledger; additive for old snapshots.
#[serde(default)]
pub processed_bytes: u64,
/// Total bytes from a complete usage snapshot, when available.
#[serde(default)]
pub total_bytes: u64,
/// Generation of the usage snapshot used for the baseline.
#[serde(default)]
pub baseline_generation: Option<u64>,
/// Whether the usage baseline is known. Missing in old snapshots means
/// indeterminate rather than a measured zero baseline.
#[serde(default)]
pub baseline_known: bool,
/// Persistent telemetry fence for counter/byte overflow or corruption.
/// It must survive a restart so a saturated snapshot is never presented as
/// a measured percentage on the next resume.
#[serde(default)]
pub counter_unknown: bool,
}
impl ResumeState {
@@ -401,8 +377,6 @@ impl ResumeState {
successful_objects: 0,
failed_objects: 0,
skipped_objects: 0,
skipped_new_versions: 0,
skipped_ilm_expired: 0,
current_bucket: None,
current_object: None,
completed_buckets: Vec::new(),
@@ -410,11 +384,6 @@ impl ResumeState {
error_message: None,
retry_count: 0,
max_retries: 3,
processed_bytes: 0,
total_bytes: 0,
baseline_generation: None,
baseline_known: false,
counter_unknown: false,
}
}
@@ -443,39 +412,6 @@ impl ResumeState {
self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
}
pub fn update_progress_with_bytes(
&mut self,
processed: u64,
successful: u64,
failed: u64,
skipped: u64,
processed_bytes: u64,
) {
self.update_progress(processed, successful, failed, skipped);
self.processed_bytes = processed_bytes;
}
pub fn set_skipped_version_counts(&mut self, new_versions: u64, ilm_expired: u64) {
self.skipped_new_versions = new_versions;
self.skipped_ilm_expired = ilm_expired;
self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
}
pub fn set_progress_baseline(&mut self, total_objects: u64, total_bytes: u64, generation: Option<u64>) {
self.total_objects = total_objects;
self.total_bytes = total_bytes;
self.baseline_generation = generation;
// This method is called only after a complete usage snapshot has been
// validated. A complete but empty snapshot is still a known baseline.
self.baseline_known = true;
self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
}
pub fn mark_counter_unknown(&mut self) {
self.counter_unknown = true;
self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
}
pub fn set_current_item(&mut self, bucket: Option<String>, object: Option<String>) {
self.current_bucket = bucket;
self.current_object = object;
@@ -501,7 +437,6 @@ impl ResumeState {
if let Some(pos) = self.pending_buckets.iter().position(|b| b == bucket) {
self.pending_buckets.remove(pos);
}
self.resume_cursor = None;
self.last_update = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
}
@@ -519,10 +454,6 @@ impl ResumeState {
self.successful_objects = 0;
self.failed_objects = 0;
self.skipped_objects = 0;
self.skipped_new_versions = 0;
self.skipped_ilm_expired = 0;
self.processed_bytes = 0;
self.counter_unknown = false;
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.
@@ -545,28 +476,14 @@ impl ResumeState {
}
pub fn get_progress_percentage(&self) -> f64 {
if self.completed {
return 100.0;
}
if self.counter_unknown {
return 0.0;
}
if !self.baseline_known {
return 0.0;
}
if self.total_bytes > 0 {
return ((self.processed_bytes as f64 / self.total_bytes as f64) * 100.0).min(99.999);
}
if self.total_objects == 0 {
return 0.0;
}
((self.processed_objects as f64 / self.total_objects as f64) * 100.0).min(99.999)
(self.processed_objects as f64 / self.total_objects as f64) * 100.0
}
pub fn get_success_rate(&self) -> f64 {
let Some(total) = self.successful_objects.checked_add(self.failed_objects) else {
return 0.0;
};
let total = self.successful_objects + self.failed_objects;
if total == 0 {
return 0.0;
}
@@ -837,14 +754,6 @@ impl ResumeManager {
state.successful_objects = 0;
state.failed_objects = 0;
state.skipped_objects = 0;
state.skipped_new_versions = 0;
state.skipped_ilm_expired = 0;
state.processed_bytes = 0;
state.total_objects = 0;
state.total_bytes = 0;
state.baseline_generation = None;
state.baseline_known = false;
state.counter_unknown = false;
state.completed = false;
state.completed_buckets.clear();
state.schema_version = CURRENT_RESUME_SCHEMA;
@@ -929,41 +838,6 @@ impl ResumeManager {
self.save_state_throttled().await
}
pub async fn update_progress_with_bytes(
&self,
processed: u64,
successful: u64,
failed: u64,
skipped: u64,
processed_bytes: u64,
) -> Result<()> {
let mut state = self.state.write().await;
state.update_progress_with_bytes(processed, successful, failed, skipped, processed_bytes);
drop(state);
self.save_state_throttled().await
}
pub async fn set_progress_baseline(&self, total_objects: u64, total_bytes: u64, generation: Option<u64>) -> Result<()> {
let mut state = self.state.write().await;
state.set_progress_baseline(total_objects, total_bytes, generation);
drop(state);
self.save_state_throttled().await
}
pub async fn mark_counter_unknown(&self) -> Result<()> {
let mut state = self.state.write().await;
state.mark_counter_unknown();
drop(state);
self.save_state().await
}
pub async fn set_skipped_version_counts(&self, new_versions: u64, ilm_expired: u64) -> Result<()> {
let mut state = self.state.write().await;
state.set_skipped_version_counts(new_versions, ilm_expired);
drop(state);
self.save_state_throttled().await
}
/// Set current item. Called once per healed object, so persistence is
/// throttled: the in-memory state always updates, but the snapshot is only
/// written every `PERSIST_EVERY_MUTATIONS` calls or `PERSIST_INTERVAL`.
@@ -1008,7 +882,7 @@ impl ResumeManager {
let mut state = self.state.write().await;
state.complete_bucket(bucket);
drop(state);
self.save_state().await
self.save_state_throttled().await
}
/// mark task completed
+1 -178
View File
@@ -34,13 +34,6 @@ const EVENT_HEAL_CHECKPOINT_STATE: &str = "heal_checkpoint_state";
/// to the new `compose_key` identities, so a stale checkpoint is discarded.
pub(super) const CURRENT_CHECKPOINT_SCHEMA: u32 = 5;
#[derive(Debug, Clone, Copy)]
pub enum CheckpointObjectOutcome {
Processed,
Failed,
Skipped,
}
/// resume checkpoint
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResumeCheckpoint {
@@ -64,30 +57,6 @@ pub struct ResumeCheckpoint {
pub failed_objects: HashSet<String>,
/// skipped objects
pub skipped_objects: HashSet<String>,
/// Aggregate object ledger counters restored alongside the dedup sets.
#[serde(default)]
pub successful_objects: u64,
#[serde(default)]
pub failed_object_count: u64,
#[serde(default)]
pub skipped_object_count: u64,
#[serde(default)]
pub skipped_new_versions: u64,
#[serde(default)]
pub skipped_ilm_expired: u64,
#[serde(default)]
pub processed_bytes: u64,
#[serde(default)]
pub total_objects: u64,
#[serde(default)]
pub total_bytes: u64,
#[serde(default)]
pub baseline_generation: Option<u64>,
#[serde(default)]
pub baseline_known: bool,
/// Persistent telemetry fence for counter/byte overflow or corruption.
#[serde(default)]
pub counter_unknown: bool,
}
impl ResumeCheckpoint {
@@ -101,17 +70,6 @@ impl ResumeCheckpoint {
processed_objects: HashSet::new(),
failed_objects: HashSet::new(),
skipped_objects: HashSet::new(),
successful_objects: 0,
failed_object_count: 0,
skipped_object_count: 0,
skipped_new_versions: 0,
skipped_ilm_expired: 0,
processed_bytes: 0,
total_objects: 0,
total_bytes: 0,
baseline_generation: None,
baseline_known: false,
counter_unknown: false,
}
}
@@ -133,34 +91,6 @@ impl ResumeCheckpoint {
self.skipped_objects.insert(object);
}
pub fn update_progress(&mut self, successful: u64, failed: u64, skipped: u64, bytes: u64) {
self.successful_objects = successful;
self.failed_object_count = failed;
self.skipped_object_count = skipped;
self.processed_bytes = bytes;
}
pub fn set_progress_baseline(&mut self, total_objects: u64, total_bytes: u64, generation: Option<u64>) {
self.total_objects = total_objects;
self.total_bytes = total_bytes;
self.baseline_generation = generation;
// The caller has already validated that this is a complete snapshot;
// preserve the distinction between a known empty scope and an old
// checkpoint that omitted all baseline fields.
self.baseline_known = true;
}
pub fn mark_counter_unknown(&mut self) {
self.counter_unknown = true;
self.checkpoint_time = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
}
pub fn set_skipped_version_counts(&mut self, new_versions: u64, ilm_expired: u64) {
self.skipped_new_versions = new_versions;
self.skipped_ilm_expired = ilm_expired;
self.checkpoint_time = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
}
/// Advance past a fully-processed page: objects below `object_index` are
/// skipped by position on resume, so the per-object sets no longer need
/// their entries and would otherwise grow with the whole bucket.
@@ -177,17 +107,6 @@ impl ResumeCheckpoint {
self.update_position(0, 0);
self.processed_objects.clear();
self.skipped_objects.clear();
self.successful_objects = 0;
self.failed_object_count = 0;
self.skipped_object_count = 0;
self.skipped_new_versions = 0;
self.skipped_ilm_expired = 0;
self.processed_bytes = 0;
self.total_objects = 0;
self.total_bytes = 0;
self.baseline_generation = None;
self.baseline_known = false;
self.counter_unknown = false;
self.failed_objects.clear();
}
}
@@ -266,17 +185,6 @@ impl CheckpointManager {
checkpoint.processed_objects.clear();
checkpoint.failed_objects.clear();
checkpoint.skipped_objects.clear();
checkpoint.successful_objects = 0;
checkpoint.failed_object_count = 0;
checkpoint.skipped_object_count = 0;
checkpoint.skipped_new_versions = 0;
checkpoint.skipped_ilm_expired = 0;
checkpoint.processed_bytes = 0;
checkpoint.total_objects = 0;
checkpoint.total_bytes = 0;
checkpoint.baseline_generation = None;
checkpoint.baseline_known = false;
checkpoint.counter_unknown = false;
checkpoint.current_bucket_index = 0;
checkpoint.current_object_index = 0;
checkpoint.schema_version = CURRENT_CHECKPOINT_SCHEMA;
@@ -317,7 +225,7 @@ impl CheckpointManager {
self.save_checkpoint_throttled().await
}
/// Persist a completed page position while retaining its identities.
/// Advance past a completed page and prune the per-object sets, then persist.
pub async fn complete_page(&self, bucket_index: usize, object_index: usize) -> Result<()> {
let mut checkpoint = self.checkpoint.write().await;
checkpoint.complete_page(bucket_index, object_index);
@@ -325,35 +233,6 @@ impl CheckpointManager {
self.save_checkpoint_throttled().await
}
/// Persist the page position while retaining identities until the resume
/// cursor is durable.
pub async fn advance_page(&self, bucket_index: usize, object_index: usize) -> Result<()> {
let mut checkpoint = self.checkpoint.write().await;
checkpoint.update_position(bucket_index, object_index);
drop(checkpoint);
self.save_checkpoint().await
}
/// Remove the previous page's dedup identities only after its resume cursor
/// has been durably exposed.
pub async fn prune_completed_page(&self) -> Result<()> {
let mut checkpoint = self.checkpoint.write().await;
checkpoint.processed_objects.clear();
checkpoint.skipped_objects.clear();
checkpoint.failed_objects.clear();
drop(checkpoint);
self.save_checkpoint().await
}
/// Advance to the next bucket and clear the final page identities after the
/// resume state has durably recorded the completed bucket.
pub async fn complete_bucket(&self, next_bucket_index: usize) -> Result<()> {
let mut checkpoint = self.checkpoint.write().await;
checkpoint.complete_page(next_bucket_index, 0);
drop(checkpoint);
self.save_checkpoint().await
}
/// Reset the checkpoint to the start of the scan for a retry, then persist.
pub async fn reset_for_retry(&self) -> Result<()> {
let mut checkpoint = self.checkpoint.write().await;
@@ -388,62 +267,6 @@ impl CheckpointManager {
self.save_checkpoint_if_due().await
}
/// Atomically persist an object's dedup identity with its aggregate result.
pub async fn record_object_outcome(
&self,
object: String,
outcome: CheckpointObjectOutcome,
successful: u64,
failed: u64,
skipped: u64,
bytes: u64,
skipped_new_versions: u64,
skipped_ilm_expired: u64,
counter_unknown: bool,
) -> Result<()> {
let mut checkpoint = self.checkpoint.write().await;
match outcome {
CheckpointObjectOutcome::Processed => checkpoint.add_processed_object(object),
CheckpointObjectOutcome::Failed => checkpoint.add_failed_object(object),
CheckpointObjectOutcome::Skipped => checkpoint.add_skipped_object(object),
}
checkpoint.update_progress(successful, failed, skipped, bytes);
checkpoint.set_skipped_version_counts(skipped_new_versions, skipped_ilm_expired);
if counter_unknown {
checkpoint.mark_counter_unknown();
}
drop(checkpoint);
self.save_checkpoint_if_due().await
}
pub async fn update_progress(&self, successful: u64, failed: u64, skipped: u64, bytes: u64) -> Result<()> {
let mut checkpoint = self.checkpoint.write().await;
checkpoint.update_progress(successful, failed, skipped, bytes);
drop(checkpoint);
self.save_checkpoint_if_due().await
}
pub async fn set_progress_baseline(&self, total_objects: u64, total_bytes: u64, generation: Option<u64>) -> Result<()> {
let mut checkpoint = self.checkpoint.write().await;
checkpoint.set_progress_baseline(total_objects, total_bytes, generation);
drop(checkpoint);
self.save_checkpoint_throttled().await
}
pub async fn mark_counter_unknown(&self) -> Result<()> {
let mut checkpoint = self.checkpoint.write().await;
checkpoint.mark_counter_unknown();
drop(checkpoint);
self.save_checkpoint().await
}
pub async fn set_skipped_version_counts(&self, new_versions: u64, ilm_expired: u64) -> Result<()> {
let mut checkpoint = self.checkpoint.write().await;
checkpoint.set_skipped_version_counts(new_versions, ilm_expired);
drop(checkpoint);
self.save_checkpoint_throttled().await
}
async fn save_checkpoint_if_due(&self) -> Result<()> {
let should_save = self.throttle.lock().map(|mut throttle| throttle.record()).unwrap_or(true);
if !should_save {
-149
View File
@@ -1296,7 +1296,6 @@ async fn test_resume_state_progress() {
assert_eq!(progress, 0.0); // total_objects is 0
state.total_objects = 100;
state.baseline_known = true;
let progress = state.get_progress_percentage();
assert_eq!(progress, 10.0);
}
@@ -1476,40 +1475,6 @@ fn test_checkpoint_object_sets_dedupe_and_prune() {
assert!(checkpoint.failed_objects.is_empty());
}
#[tokio::test]
async fn checkpoint_page_commit_keeps_ledger_until_cursor_is_durable() {
let (_temp_dir, disk) = schema_test_disk().await;
let task_id = ResumeUtils::generate_task_id();
let checkpoint = CheckpointManager::new(disk.clone(), task_id.clone()).await.unwrap();
checkpoint
.record_object_outcome(
"bucket/object:v1".to_string(),
CheckpointObjectOutcome::Processed,
1,
0,
0,
128,
0,
0,
false,
)
.await
.unwrap();
checkpoint.advance_page(0, 1).await.unwrap();
let reloaded = CheckpointManager::load_from_disk(disk.clone(), &task_id).await.unwrap();
let snapshot = reloaded.get_checkpoint().await;
assert_eq!(snapshot.current_object_index, 1);
assert_eq!(snapshot.successful_objects, 1);
assert_eq!(snapshot.processed_bytes, 128);
assert!(snapshot.processed_objects.contains("bucket/object:v1"));
checkpoint.prune_completed_page().await.unwrap();
let reloaded = CheckpointManager::load_from_disk(disk, &task_id).await.unwrap();
assert!(reloaded.get_checkpoint().await.processed_objects.is_empty());
}
#[test]
fn test_checkpoint_loads_legacy_vec_format() {
// Checkpoints written before the HashSet migration stored the object
@@ -1674,120 +1639,6 @@ async fn current_normal_resume_schema_preserves_progress() {
temp_dir.close().expect("remove schema test directory");
}
#[test]
fn progress_checkpoint_restores_bytes_and_generation() {
let mut checkpoint = ResumeCheckpoint::new("progress-checkpoint".to_string());
checkpoint.set_progress_baseline(9, 4096, Some(77));
checkpoint.update_progress(4, 1, 2, 2048);
checkpoint.set_skipped_version_counts(3, 1);
checkpoint.mark_counter_unknown();
let restored: ResumeCheckpoint =
serde_json::from_slice(&serde_json::to_vec(&checkpoint).expect("serialize checkpoint")).expect("deserialize checkpoint");
assert_eq!(restored.processed_bytes, 2048);
assert_eq!(restored.total_objects, 9);
assert_eq!(restored.total_bytes, 4096);
assert_eq!(restored.baseline_generation, Some(77));
assert!(restored.baseline_known);
assert_eq!(restored.skipped_new_versions, 3);
assert_eq!(restored.skipped_ilm_expired, 1);
assert!(restored.counter_unknown);
}
#[test]
fn old_progress_schema_migrates_missing_fields_to_unknown() {
let state = ResumeState::new(
"legacy-progress".to_string(),
"erasure_set".to_string(),
"pool_0_set_0".to_string(),
Vec::new(),
);
let mut value = serde_json::to_value(state).expect("serialize legacy-compatible state");
let object = value.as_object_mut().expect("state must be an object");
for field in [
"processed_bytes",
"total_bytes",
"baseline_generation",
"baseline_known",
"skipped_new_versions",
"skipped_ilm_expired",
] {
object.remove(field);
}
object.insert("total_objects".to_string(), serde_json::json!(10));
object.insert("processed_objects".to_string(), serde_json::json!(5));
let restored: ResumeState = serde_json::from_value(value).expect("deserialize old progress state");
assert_eq!(restored.processed_bytes, 0);
assert_eq!(restored.total_bytes, 0);
assert_eq!(restored.baseline_generation, None);
assert!(!restored.baseline_known, "missing baseline must remain unknown");
assert_eq!(restored.get_progress_percentage(), 0.0);
assert_eq!(restored.skipped_new_versions, 0);
assert_eq!(restored.skipped_ilm_expired, 0);
}
#[test]
fn progress_counter_unknown_survives_resume_round_trip() {
let mut state = ResumeState::new(
"overflow-progress".to_string(),
"erasure_set".to_string(),
"pool_0_set_0".to_string(),
Vec::new(),
);
state.mark_counter_unknown();
let restored: ResumeState =
serde_json::from_slice(&serde_json::to_vec(&state).expect("serialize resume state")).expect("deserialize resume state");
assert!(restored.counter_unknown);
}
#[tokio::test]
async fn checkpoint_progress_survives_a_torn_resume_summary_write() {
let (_temp_dir, disk) = schema_test_disk().await;
let task_id = ResumeUtils::generate_task_id();
let _resume = ResumeManager::new(
disk.clone(),
task_id.clone(),
"erasure_set".to_string(),
"pool_0_set_0".to_string(),
vec!["bucket".to_string()],
)
.await
.expect("resume state should persist");
let checkpoint = CheckpointManager::new(disk.clone(), task_id.clone())
.await
.expect("checkpoint should persist");
// This is the ordering used by the erasure-set loop: the checkpoint is
// durable before the summary write. Stop here to model a crash in the
// inter-store window and verify that the recovery authority retains the
// telemetry fence and bytes.
checkpoint
.update_progress(3, 0, 0, 1024)
.await
.expect("checkpoint progress should persist");
checkpoint.mark_counter_unknown().await.expect("unknown fence should persist");
checkpoint
.update_position(0, 3)
.await
.expect("checkpoint position should persist");
let restored_checkpoint = CheckpointManager::load_from_disk(disk.clone(), &task_id)
.await
.expect("checkpoint should reload")
.get_checkpoint()
.await;
let restored_resume = ResumeManager::load_from_disk(disk, &task_id)
.await
.expect("resume summary should reload")
.get_state()
.await;
assert!(restored_checkpoint.counter_unknown);
assert_eq!(restored_checkpoint.processed_bytes, 1024);
assert_eq!(restored_checkpoint.current_object_index, 3);
assert!(!restored_resume.counter_unknown, "summary is intentionally the torn/older store");
}
#[tokio::test]
async fn future_resume_and_checkpoint_schemas_are_rejected() {
let (temp_dir, disk) = schema_test_disk().await;
+2 -26
View File
@@ -19,8 +19,6 @@ 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::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::sync::Arc;
use tracing::{debug, error, warn};
@@ -36,9 +34,6 @@ pub use super::{HealObjectInfo, HealObjectOptions, HealPutObjReader};
pub struct HealBucketUsageBaseline {
pub objects_count: u64,
pub bytes: u64,
/// Stable identity of the validated usage snapshot and selected scope.
/// `None` is retained for test/legacy providers that cannot expose one.
pub generation: Option<u64>,
}
pub struct HealLifecycleExpiryContext {
@@ -790,30 +785,11 @@ impl HealStorageAPI for ECStoreHealStorage {
let mut baseline = HealBucketUsageBaseline::default();
for bucket in buckets {
if let Some(usage) = info.buckets_usage.get(bucket) {
baseline.objects_count = match baseline.objects_count.checked_add(usage.objects_count) {
Some(total) => total,
// A corrupt/overflowing usage snapshot is not a usable
// denominator. Leave progress indeterminate instead of
// turning saturation into a plausible percentage.
None => return Ok(None),
};
baseline.bytes = match baseline.bytes.checked_add(usage.size) {
Some(total) => total,
None => return Ok(None),
};
baseline.objects_count = baseline.objects_count.saturating_add(usage.objects_count);
baseline.bytes = baseline.bytes.saturating_add(usage.size);
}
}
let identity = info.snapshot_identity();
let mut hasher = DefaultHasher::new();
identity.last_update.hash(&mut hasher);
identity.scanner_cycle.hash(&mut hasher);
identity.scanner_epoch.hash(&mut hasher);
let mut scope = buckets.to_vec();
scope.sort_unstable();
scope.hash(&mut hasher);
baseline.generation = Some(hasher.finish());
Ok(Some(baseline))
}
+3 -7
View File
@@ -649,7 +649,7 @@ impl HealTask {
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("skipped: {bucket}/{object}")));
progress.update_stage(1, 1);
progress.update_progress(0, 1, 0, 0);
Ok(())
}
@@ -733,7 +733,7 @@ impl HealTask {
"Heal object skipped for data usage cache after transient error"
);
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
true
}
@@ -757,7 +757,7 @@ impl HealTask {
);
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("skipped: {bucket}/{object}")));
progress.update_stage(4, 4);
progress.update_progress(4, 4, 0, 0);
true
}
@@ -831,10 +831,6 @@ impl HealTask {
match &result {
Ok(_) => {
// A stage can reach its final step before the durable resume
// ledger and cleanup fences commit. Publish terminal 100 only
// after the enclosing operation has returned success.
self.progress.write().await.mark_completed();
let mut status = self.status.write().await;
*status = HealTaskStatus::Completed;
demote_to_debug_when!(self.heal_type.is_per_object(), info, target: "rustfs::heal::task", {
+16 -46
View File
@@ -13,7 +13,6 @@
// limitations under the License.
/// bucket/cluster/prefix heal: the recursive bucket-objects sweep and the erasure-set usage baseline
use super::*;
use crate::heal::progress::{add_bytes, increment_counter};
impl HealTask {
pub(super) async fn heal_bucket(&self, bucket: &str) -> Result<()> {
@@ -33,7 +32,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("bucket: {bucket}")));
progress.update_stage(0, 3);
progress.update_progress(0, 3, 0, 0);
}
// Step 1: Check if bucket exists
@@ -67,7 +66,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_stage(1, 3);
progress.update_progress(1, 3, 0, 0);
}
// Step 2: Perform bucket heal using ecstore
@@ -123,7 +122,7 @@ impl HealTask {
if !self.options.recursive {
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
}
Ok(())
}
@@ -143,7 +142,7 @@ impl HealTask {
);
{
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
}
Err(Error::TaskExecutionFailed {
message: format!("Failed to heal bucket {bucket}: {e}"),
@@ -246,7 +245,6 @@ impl HealTask {
let mut scanned = 0u64;
let mut healed = 0u64;
let mut failed = 0u64;
let mut skipped = 0u64;
let mut retryable_failed = 0u64;
let mut permanent_failed = 0u64;
let mut bytes = 0u64;
@@ -288,14 +286,16 @@ impl HealTask {
let mut retry = Vec::with_capacity(pending.len());
for item in pending {
self.check_control_flags().await?;
let mut telemetry_unknown = false;
let object = item.name.as_str();
if retry_attempt == 0 {
scanned = scanned.saturating_add(1);
}
{
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("{bucket}/{object}")));
progress.update_progress(scanned, healed, failed, bytes);
}
let mut terminal_outcome = true;
let error = match self
.await_with_control(
self.storage
@@ -304,13 +304,13 @@ impl HealTask {
.await
{
Ok((result, None)) => {
telemetry_unknown |= !increment_counter(&mut healed);
telemetry_unknown |= !add_bytes(&mut bytes, u64::try_from(result.object_size).unwrap_or(u64::MAX));
healed = healed.saturating_add(1);
bytes = bytes.saturating_add(u64::try_from(result.object_size).unwrap_or_default());
self.record_result_item(result).await;
None
}
Ok((_, Some(err))) if is_missing_object_dir_heal_result(object, &err) => {
telemetry_unknown |= !increment_counter(&mut healed);
healed = healed.saturating_add(1);
debug!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
@@ -329,7 +329,6 @@ impl HealTask {
if let Some(err) = error {
if Self::should_skip_data_usage_cache_heal_error(bucket, object, &err) {
telemetry_unknown |= !increment_counter(&mut skipped);
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
@@ -343,7 +342,6 @@ impl HealTask {
"Heal bucket object repair skipped due to transient metadata error"
);
} else if err.is_recoverable_heal() && retry_attempt < MAX_BUCKET_OBJECT_HEAL_RETRIES {
terminal_outcome = false;
debug!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
@@ -359,7 +357,7 @@ impl HealTask {
);
retry.push(item);
} else {
telemetry_unknown |= !increment_counter(&mut failed);
failed = failed.saturating_add(1);
if err.is_recoverable_heal() {
retryable_failed = retryable_failed.saturating_add(1);
} else {
@@ -385,19 +383,8 @@ impl HealTask {
}
}
if terminal_outcome {
telemetry_unknown |= !increment_counter(&mut scanned);
}
if !terminal_outcome {
continue;
}
let mut progress = self.progress.write().await;
progress.update_object_progress(scanned, healed, failed, skipped, bytes);
if telemetry_unknown {
progress.mark_unknown();
}
progress.update_progress(scanned, healed, failed, bytes);
}
pending = retry;
retry_attempt = retry_attempt.saturating_add(1);
@@ -444,7 +431,7 @@ impl HealTask {
Ok(())
}
pub(super) async fn apply_erasure_set_usage_baseline(&self, buckets: &[String], set_disk_id: &str) -> Result<()> {
pub(super) async fn apply_erasure_set_usage_baseline(&self, buckets: &[String]) -> Result<()> {
let baseline = match self
.await_with_control(self.storage.erasure_set_usage_baseline(buckets))
.await
@@ -455,26 +442,9 @@ impl HealTask {
Err(_) => return Ok(()),
};
let HealBucketUsageBaseline {
objects_count,
bytes,
generation,
} = baseline;
let generation = generation.map(|snapshot_generation| {
use std::hash::{Hash, Hasher};
let mut hasher = std::collections::hash_map::DefaultHasher::new();
snapshot_generation.hash(&mut hasher);
set_disk_id.hash(&mut hasher);
self.options.pool_index.hash(&mut hasher);
self.options.set_index.hash(&mut hasher);
hasher.finish()
});
let HealBucketUsageBaseline { objects_count, bytes } = baseline;
let mut progress = self.progress.write().await;
if let Some(generation) = generation {
progress.set_total_baseline_with_generation(objects_count, bytes, generation);
} else {
progress.set_total_baseline(objects_count, bytes);
}
progress.set_total_baseline(objects_count, bytes);
Ok(())
}
}
+10 -8
View File
@@ -32,7 +32,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("erasure_set: {} ({} buckets)", set_disk_id, buckets.len())));
progress.update_stage(0, 4);
progress.update_progress(0, 4, 0, 0);
}
let is_auto_replacement = matches!(self.source, HealRequestSource::AutoHeal) && !self.heal_endpoints.is_empty();
@@ -158,7 +158,7 @@ impl HealTask {
None
};
self.apply_erasure_set_usage_baseline(&buckets, &set_disk_id).await?;
self.apply_erasure_set_usage_baseline(&buckets).await?;
let healing_marker = format!("{set_disk_id}:{}", self.id);
if let Some((disk, resume_manager, _)) = replacement_resume.as_ref() {
@@ -244,7 +244,7 @@ impl HealTask {
);
{
let mut progress = self.progress.write().await;
progress.update_stage(4, 4);
progress.update_progress(4, 4, 0, 0);
}
return Err(Error::TaskExecutionFailed {
message: format!("Failed to heal disk format for {set_disk_id}: {e}"),
@@ -297,7 +297,7 @@ impl HealTask {
);
{
let mut progress = self.progress.write().await;
progress.update_stage(4, 4);
progress.update_progress(4, 4, 0, 0);
}
return Err(Error::TaskExecutionFailed {
message: format!("Failed to heal disk format for {set_disk_id}: {e}"),
@@ -307,7 +307,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_stage(1, 4);
progress.update_progress(1, 4, 0, 0);
}
// The rebuilt disks are formatted now: mark them as healing so
@@ -336,7 +336,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_stage(2, 4);
progress.update_progress(2, 4, 0, 0);
}
// Step 3: Heal bucket structure
@@ -420,7 +420,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_stage(3, 4);
progress.update_progress(3, 4, 0, 0);
}
// Step 4: Execute erasure set heal with resume
@@ -463,7 +463,9 @@ impl HealTask {
};
{
self.progress.write().await.update_stage(4, 4);
let mut progress = self.progress.write().await;
let bytes_processed = progress.bytes_processed;
progress.update_progress(4, 4, 0, bytes_processed);
}
match result {
+10 -10
View File
@@ -32,7 +32,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("metadata: {bucket}/{object}")));
progress.update_stage(0, 3);
progress.update_progress(0, 3, 0, 0);
}
// Step 1: Check if object exists
@@ -74,7 +74,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_stage(1, 3);
progress.update_progress(1, 3, 0, 0);
}
// Step 2: Perform metadata heal using ecstore
@@ -122,7 +122,7 @@ impl HealTask {
);
{
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
}
return Err(Error::TaskExecutionFailed {
message: format!("Failed to heal metadata {bucket}/{object}: {e}"),
@@ -145,7 +145,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
}
self.record_result_item(result).await;
Ok(())
@@ -167,7 +167,7 @@ impl HealTask {
);
{
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
}
Err(Error::TaskExecutionFailed {
message: format!("Failed to heal metadata {bucket}/{object}: {e}"),
@@ -194,7 +194,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("ec_decode: {bucket}/{object}")));
progress.update_stage(0, 3);
progress.update_progress(0, 3, 0, 0);
}
// Step 1: Check if object exists
@@ -236,7 +236,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_stage(1, 3);
progress.update_progress(1, 3, 0, 0);
}
// Step 2: Perform EC decode heal using ecstore
@@ -284,7 +284,7 @@ impl HealTask {
);
{
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
}
return Err(Error::TaskExecutionFailed {
message: format!("Failed to heal EC decode {bucket}/{object}: {e}"),
@@ -309,7 +309,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_object_progress(1, 1, 0, 0, object_size);
progress.update_progress(3, 3, 0, object_size);
}
self.record_result_item(result).await;
Ok(())
@@ -331,7 +331,7 @@ impl HealTask {
);
{
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
}
Err(Error::TaskExecutionFailed {
message: format!("Failed to heal EC decode {bucket}/{object}: {e}"),
+8 -8
View File
@@ -36,7 +36,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.set_current_object(Some(format!("{bucket}/{object}")));
progress.update_stage(0, 4);
progress.update_progress(0, 4, 0, 0);
}
// Step 1: Check if object exists and get metadata
@@ -132,7 +132,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_stage(1, 3);
progress.update_progress(1, 3, 0, 0);
}
// Step 2: directly call ecstore to perform heal
@@ -187,7 +187,7 @@ impl HealTask {
);
{
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
}
return Ok(());
}
@@ -207,7 +207,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
}
if Self::should_return_typed_heal_error(&e) {
@@ -249,7 +249,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_object_progress(1, 1, 0, 0, object_size);
progress.update_progress(3, 3, 0, object_size);
}
self.record_result_item(result).await;
Ok(())
@@ -275,7 +275,7 @@ impl HealTask {
);
{
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
}
return Ok(());
}
@@ -295,7 +295,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_stage(3, 3);
progress.update_progress(3, 3, 0, 0);
}
if Self::should_return_typed_heal_error(&e) {
@@ -414,7 +414,7 @@ impl HealTask {
{
let mut progress = self.progress.write().await;
progress.update_object_progress(1, 1, 0, 0, object_size);
progress.update_progress(4, 4, 0, object_size);
}
self.record_result_item(result).await;
Ok(())
-3
View File
@@ -2096,7 +2096,6 @@ async fn erasure_set_heal_applies_usage_baseline_to_progress() {
usage_baseline: Mutex::new(Some(HealBucketUsageBaseline {
objects_count: 10,
bytes: 8,
generation: Some(1),
})),
..Default::default()
});
@@ -2120,8 +2119,6 @@ async fn erasure_set_heal_applies_usage_baseline_to_progress() {
let progress = task.get_progress().await;
assert_eq!(progress.objects_total_count, 10);
assert_eq!(progress.objects_total_size, 8);
assert!(progress.baseline_generation.is_some());
assert!(progress.baseline_known);
assert_eq!(progress.bytes_processed, 2);
assert!((progress.progress_percentage - 25.0).abs() < 0.001);
}
+2 -7
View File
@@ -60,9 +60,7 @@ pub const REPLICATION_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[
"Destination.ReplicationTime",
];
// v2: disableProxy moved from unsupported to writable (per-target read-proxy
// opt-out is accepted by set-remote-target and the `proxy` update op).
pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 2;
pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 1;
pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
"sourcebucket",
@@ -85,12 +83,9 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[
// madmin default of 60s); the per-target health-check interval is not
// yet applied — the heartbeat keeps its global env-configured interval.
"healthCheckDuration",
// Per-target read-proxy opt-out, consumed by the proxy-target selector
// (contract v2; previously only importable via MinIO bucket-targets.json).
"disableProxy",
];
pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["edge", "edgeSyncBeforeExpiry"];
pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["disableProxy", "edge", "edgeSyncBeforeExpiry"];
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct ObjectOpts {
+50 -5
View File
@@ -29,7 +29,8 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
pub use rustfs_data_usage::{
AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path, prefix_usage_in_cache,
PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeReconciliationEntry, SizeReconciliationScope, SizeSummary,
TierStats, hash_path, prefix_usage_in_cache,
};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use tokio::time::{Duration, Instant, sleep, timeout};
@@ -192,6 +193,10 @@ const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
pub trait ScannerSizeSummaryExt {
/// Fold one object's contribution into the summary, including its tier.
fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64);
/// Fold counters and physical tier usage for an object whose metadata is
/// valid but whose logical size is currently unavailable. Logical totals
/// stay unchanged.
fn actions_accounting_unknown(&mut self, oi: &ObjectInfo);
}
impl ScannerSizeSummaryExt for SizeSummary {
@@ -225,6 +230,34 @@ impl ScannerSizeSummaryExt for SizeSummary {
});
}
}
fn actions_accounting_unknown(&mut self, oi: &ObjectInfo) {
if oi.delete_marker {
self.delete_markers = self.delete_markers.saturating_add(1);
return;
}
if oi.version_id.is_some_and(|v| !v.is_nil()) {
self.versions = self.versions.saturating_add(1);
}
if oi.transitioned_object.free_version {
return;
}
let tier = if oi.transitioned_object.status == TRANSITION_COMPLETE {
oi.transitioned_object.tier.clone()
} else {
oi.storage_class.clone().unwrap_or_else(|| storageclass::STANDARD.to_string())
};
if let Some(tier_stats) = self.tier_stats.get_mut(&tier) {
*tier_stats = tier_stats.add(&TierStats {
total_size: u64::try_from(oi.size).unwrap_or(0),
num_versions: 1,
num_objects: u64::from(oi.is_latest),
});
}
}
}
// ===== Cache-related data structures =====
@@ -344,6 +377,10 @@ pub struct DataUsageCacheInfo {
pub scan_plan_digest: Option<DataUsageScanPlanDigest>,
#[serde(default)]
pub cache_key_format: u16,
/// Bounded durable debts for versions whose logical size was not trusted.
/// The map key is an identity key, never a user-controlled metric label.
#[serde(default)]
pub size_reconciliation: HashMap<String, SizeReconciliationEntry>,
}
impl Serialize for DataUsageCacheInfo {
@@ -353,7 +390,8 @@ impl Serialize for DataUsageCacheInfo {
{
// Keep this metadata map-encoded so older readers can ignore fields
// appended by newer scanner versions during rolling upgrades.
let mut state = serializer.serialize_map(Some(16))?;
let field_count = 16 + usize::from(!self.size_reconciliation.is_empty());
let mut state = serializer.serialize_map(Some(field_count))?;
state.serialize_entry("name", &self.name)?;
state.serialize_entry("next_cycle", &self.next_cycle)?;
state.serialize_entry("leader_epoch", &self.leader_epoch)?;
@@ -370,6 +408,9 @@ impl Serialize for DataUsageCacheInfo {
state.serialize_entry("snapshot_complete", &self.snapshot_complete)?;
state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?;
state.serialize_entry("cache_key_format", &self.cache_key_format)?;
if !self.size_reconciliation.is_empty() {
state.serialize_entry("size_reconciliation", &self.size_reconciliation)?;
}
state.end()
}
}
@@ -428,14 +469,18 @@ impl DataUsageCache {
self.checked_flatten(name).is_some()
});
if !reusable {
let pending_heals = if self.info.name == name {
std::mem::take(&mut self.info.pending_heals)
let (pending_heals, size_reconciliation) = if self.info.name == name {
(
std::mem::take(&mut self.info.pending_heals),
std::mem::take(&mut self.info.size_reconciliation),
)
} else {
Vec::new()
(Vec::new(), HashMap::new())
};
*self = Self::default();
self.info.name = name.to_string();
self.info.pending_heals = pending_heals;
self.info.size_reconciliation = size_reconciliation;
}
self.info.next_cycle = next_cycle;
@@ -673,6 +673,34 @@ fn size_summary_actions_accounting_accumulates_tier_stats() {
);
}
#[test]
fn size_summary_unknown_accounting_keeps_physical_tier_and_version_only() {
let mut summary = SizeSummary::default();
summary
.tier_stats
.insert(storageclass::STANDARD.to_string(), TierStats::default());
let object = ObjectInfo {
size: 12,
storage_class: Some(storageclass::STANDARD.to_string()),
version_id: Some(uuid::Uuid::new_v4()),
is_latest: true,
..Default::default()
};
summary.actions_accounting_unknown(&object);
assert_eq!(summary.total_size, 0, "unknown logical size must not become zero or physical bytes");
assert_eq!(summary.versions, 1);
assert_eq!(
summary.tier_stats.get(storageclass::STANDARD),
Some(&TierStats {
total_size: 12,
num_versions: 1,
num_objects: 1,
})
);
}
#[test]
fn test_data_usage_entry_merge_sums_failed_objects() {
let mut left = DataUsageEntry {
@@ -1079,6 +1107,16 @@ fn data_usage_cache_prepare_for_scan_preserves_pending_heal_only_progress() {
scan_plan_digest: Some(TEST_PLAN_DIGEST),
cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
pending_heals: vec![pending_heal.clone()],
size_reconciliation: HashMap::from([(
"size-key".to_string(),
SizeReconciliationEntry {
key: "size-key".to_string(),
bucket: "bucket".to_string(),
object: "prefix/object".to_string(),
reason: "invalid_declared_size".to_string(),
..Default::default()
},
)]),
..Default::default()
},
..Default::default()
@@ -1088,6 +1126,7 @@ fn data_usage_cache_prepare_for_scan_preserves_pending_heal_only_progress() {
assert_eq!(outcome, DataUsageCachePrepareOutcome::Reused);
assert_eq!(cache.info.pending_heals, vec![pending_heal]);
assert!(cache.info.size_reconciliation.contains_key("size-key"));
assert!(cache.cache.is_empty());
assert!(!cache.info.snapshot_complete);
}
+113 -5
View File
@@ -20,8 +20,9 @@ use std::time::{Duration, Instant, SystemTime};
use crate::ReplTargetSizeSummary;
use crate::data_usage_define::{
DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageScanCheckpoint,
DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt, SizeSummary, hash_path,
DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageCacheInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap,
DataUsageScanCheckpoint, DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt,
SizeReconciliationEntry, SizeSummary, hash_path,
};
use crate::error::ScannerError;
use crate::runtime_config::{
@@ -97,6 +98,9 @@ const METRIC_SCANNER_EXCESS_FOLDERS_TOTAL: &str = "rustfs_scanner_excess_folders
const METRIC_SCANNER_PENDING_HEAL_PRUNE_TOTAL: &str = "rustfs_scanner_pending_heal_prune_total";
const METRIC_SCANNER_PENDING_HEAL_MALFORMED_TOTAL: &str = "rustfs_scanner_pending_heal_malformed_total";
const MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET: usize = 128;
const MAX_SIZE_RECONCILIATION_ENTRIES_PER_BUCKET: usize = 10_000;
const MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET: usize = 8 * 1024 * 1024;
const MAX_SIZE_RECONCILIATION_AGE_SECS: u64 = 7 * 24 * 60 * 60;
// --- scanner excess alerts as S3 notification events (rustfs/backlog#1868) --
//
@@ -364,7 +368,7 @@ impl PendingScannerAccounting<'_> {
fn apply(self, size_summary: &mut SizeSummary, cumulative_size: &mut i64, queued: bool) {
let size = if queued { self.expired_size } else { self.retained_size };
size_summary.actions_accounting(self.object, size, self.retained_size);
*cumulative_size += size;
*cumulative_size = cumulative_size.saturating_add(size);
}
}
@@ -675,6 +679,54 @@ pub struct FolderScanner {
list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
}
fn size_reconciliation_entry_bytes(entry: &SizeReconciliationEntry) -> usize {
entry.key.len()
+ entry.bucket.len()
+ entry.object.len()
+ entry.version_id.as_deref().map_or(0, str::len)
+ entry.generation.as_deref().map_or(0, str::len)
+ entry.reason.len()
+ std::mem::size_of::<u64>()
+ std::mem::size_of::<u32>()
}
fn prune_size_reconciliation(info: &mut DataUsageCacheInfo, now: u64) {
info.size_reconciliation.retain(|key, entry| {
if entry.first_seen == 0 || entry.first_seen > now {
entry.first_seen = now;
}
key == &entry.key
&& entry.key.len() <= 4096
&& entry.bucket.len() <= 512
&& entry.object.len() <= 512
&& entry.version_id.as_deref().is_none_or(|value| value.len() <= 64)
&& entry.generation.as_deref().is_none_or(|value| value.len() <= 64)
&& entry.reason.len() <= 64
&& now.saturating_sub(entry.first_seen) <= MAX_SIZE_RECONCILIATION_AGE_SECS
});
while info.size_reconciliation.len() > MAX_SIZE_RECONCILIATION_ENTRIES_PER_BUCKET
|| info
.size_reconciliation
.values()
.map(size_reconciliation_entry_bytes)
.sum::<usize>()
> MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET
{
let oldest = info
.size_reconciliation
.iter()
.min_by(|(left_key, left), (right_key, right)| {
left.first_seen.cmp(&right.first_seen).then_with(|| left_key.cmp(right_key))
})
.map(|(key, _)| key.clone());
let Some(oldest) = oldest else {
break;
};
info.size_reconciliation.remove(&oldest);
}
}
impl FolderScanner {
fn now_secs() -> u64 {
SystemTime::now()
@@ -748,6 +800,55 @@ impl FolderScanner {
}
}
/// Apply the per-object size-resolution ledger updates in one place. The
/// scanner cache is the durable boundary; both working copies are updated
/// so an incremental publication cannot lose a debt or its resolution.
fn apply_size_reconciliation(&mut self, summary: &SizeSummary) {
let now = Self::now_secs();
// Keep an unresolved identity in place while refreshing its object
// scope. This lets repeated observations increment `attempts`; only
// debts absent from the current pass are considered resolved.
let current_keys = summary
.size_reconciliation
.iter()
.map(|entry| entry.key.clone())
.collect::<HashSet<_>>();
for info in [&mut self.new_cache.info, &mut self.update_cache.info] {
prune_size_reconciliation(info, now);
if !summary.size_reconciliation_truncated {
for scope in &summary.reconciliation_scopes {
let scope_bucket = item_actions::bounded_reconciliation_field(&scope.bucket);
let scope_object = item_actions::bounded_reconciliation_field(&scope.object);
info.size_reconciliation.retain(|key, entry| {
entry.bucket != scope_bucket || entry.object != scope_object || current_keys.contains(key)
});
}
}
for incoming in &summary.size_reconciliation {
if let Some(existing) = info.size_reconciliation.get_mut(&incoming.key) {
existing.reason = incoming.reason.clone();
existing.physical_size = incoming.physical_size;
existing.generation = incoming.generation.clone();
existing.version_id = incoming.version_id.clone();
existing.attempts = existing.attempts.saturating_add(1);
continue;
}
if size_reconciliation_entry_bytes(incoming) > MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET {
continue;
}
let mut entry = incoming.clone();
entry.first_seen = now;
entry.attempts = 1;
info.size_reconciliation.insert(entry.key.clone(), entry);
}
prune_size_reconciliation(info, now);
}
}
fn record_scan_resume_hint(&mut self, folder: &str) {
self.new_cache.info.scan_resume_after = Some(folder.to_string());
self.update_cache.info.scan_resume_after = Some(folder.to_string());
@@ -1426,6 +1527,7 @@ impl FolderScanner {
abandoned_children.remove(&path_join_buf(&[&item.bucket, &item.object_path()]));
apply_scanner_size_summary(into, &sz);
self.apply_size_reconciliation(&sz);
into.objects += 1;
object_count += 1;
self.budget.record_object_scanned();
@@ -2194,6 +2296,10 @@ pub async fn scan_data_folder(
list_path_raw_options_observer: None,
};
let now = FolderScanner::now_secs();
prune_size_reconciliation(&mut scanner.new_cache.info, now);
prune_size_reconciliation(&mut scanner.update_cache.info, now);
// Check if context is cancelled
if ctx.is_cancelled() {
return Err(ScannerError::Other("Operation cancelled".to_string()));
@@ -2217,7 +2323,9 @@ pub async fn scan_data_folder(
new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
new_cache.info.last_update = Some(SystemTime::now());
new_cache.info.next_cycle = cache.info.next_cycle;
let unresolved_objects = root.failed_objects > 0 || !new_cache.info.failed_objects.is_empty();
let unresolved_objects = root.failed_objects > 0
|| !new_cache.info.failed_objects.is_empty()
|| !new_cache.info.size_reconciliation.is_empty();
new_cache.info.snapshot_complete = !unresolved_objects;
let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some();
new_cache.info.scan_resume_after = None;
@@ -2245,7 +2353,7 @@ pub async fn scan_data_folder(
if root_has_progress {
new_cache.replace_hashed(&root_hash, &None, &root);
}
if partial_cache_is_useful(&root, pending_heals_changed) {
if partial_cache_is_useful(&root, pending_heals_changed) || !new_cache.info.size_reconciliation.is_empty() {
if new_cache.root().is_some() {
new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
}
+769 -65
View File
@@ -13,6 +13,7 @@
// limitations under the License.
/// Per-object scan actions: ScannerItem, the get-size failure policy, and the heal/ILM admission helpers.
use super::*;
use sha2::{Digest as _, Sha256};
/// Cached folder information for scanning
#[derive(Clone, Debug)]
@@ -32,6 +33,259 @@ pub(super) enum GetSizeFailureAction {
HealMetadata { object: String },
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum SizeResolutionReason {
CompressedSizeUnknown,
InvalidPhysicalSize,
UnsupportedCompression,
InvalidObjectSize,
InvalidPartSize,
InvalidDeclaredSize,
SizeOverflowOrMismatch,
}
impl SizeResolutionReason {
fn as_str(self) -> &'static str {
match self {
Self::CompressedSizeUnknown => "compressed_size_unknown",
Self::InvalidPhysicalSize => "invalid_physical_size",
Self::UnsupportedCompression => "unsupported_compression",
Self::InvalidObjectSize => "invalid_object_size",
Self::InvalidPartSize => "invalid_part_size",
Self::InvalidDeclaredSize => "invalid_declared_size",
Self::SizeOverflowOrMismatch => "size_overflow_or_mismatch",
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum SizeResolution {
Known { logical: i64, physical: i64 },
Unknown { physical: i64, reason: SizeResolutionReason },
Corrupt { physical: i64, reason: SizeResolutionReason },
}
impl SizeResolution {
fn known_size(&self) -> Option<i64> {
match self {
Self::Known { logical, .. } => Some(*logical),
Self::Unknown { .. } | Self::Corrupt { .. } => None,
}
}
}
fn size_reconciliation_key(oi: &ObjectInfo, reason: SizeResolutionReason) -> String {
let version = oi
.version_id
.filter(|version| !version.is_nil())
.map(|version| version.to_string())
.unwrap_or_default();
let generation = oi
.data_dir
.filter(|generation| !generation.is_nil())
.map(|generation| generation.to_string())
.unwrap_or_default();
// Length-prefix each component so an object key containing the separator
// cannot alias another identity. S3 keys are bounded in normal operation;
// oversized persisted values use a digest so a corrupt metadata record
// cannot grow the ledger without bound.
fn component(value: &str) -> String {
const MAX_COMPONENT_LEN: usize = 512;
if value.len() <= MAX_COMPONENT_LEN {
return format!("{}:{}", value.len(), value);
}
let digest = Sha256::digest(value.as_bytes());
let digest = hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower);
format!("hash:{}:{}", value.len(), digest)
}
format!(
"{}|{}|{}|{}|{}",
component(&oi.bucket),
component(&oi.name),
component(&version),
component(&generation),
component(reason.as_str())
)
}
pub(super) fn bounded_reconciliation_field(value: &str) -> String {
const MAX_FIELD_LEN: usize = 512;
if value.len() <= MAX_FIELD_LEN {
return value.to_string();
}
let digest = hex_simd::encode_to_string(Sha256::digest(value.as_bytes()), hex_simd::AsciiCase::Lower);
let prefix_len = MAX_FIELD_LEN - 65;
let prefix = value
.char_indices()
.take_while(|(offset, ch)| offset.saturating_add(ch.len_utf8()) <= prefix_len)
.map(|(_, ch)| ch)
.collect::<String>();
format!("{}~{}", prefix, digest)
}
fn record_size_resolution(summary: &mut SizeSummary, oi: &ObjectInfo, resolution: &SizeResolution) {
match resolution {
SizeResolution::Known { .. } => {}
SizeResolution::Unknown { physical, reason } | SizeResolution::Corrupt { physical, reason } => {
summary.record_size_reconciliation(SizeReconciliationEntry {
key: size_reconciliation_key(oi, *reason),
bucket: bounded_reconciliation_field(&oi.bucket),
object: bounded_reconciliation_field(&oi.name),
version_id: oi
.version_id
.filter(|version| !version.is_nil())
.map(|version| version.to_string()),
generation: oi
.data_dir
.filter(|generation| !generation.is_nil())
.map(|generation| generation.to_string()),
reason: reason.as_str().to_string(),
physical_size: u64::try_from(*physical).ok(),
first_seen: 0,
attempts: 0,
});
}
}
}
/// Resolve the size metadata once at the scanner trust boundary. A compressed
/// -1 sentinel is valid legacy metadata, but it cannot participate in normal
/// logical-size accounting or size-filtered lifecycle rules.
pub(super) fn resolve_size(oi: &ObjectInfo) -> SizeResolution {
let physical = oi.size;
if physical < 0 {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidPhysicalSize,
};
}
let compressed = match oi.compression_read_plan() {
Ok((_, _, compressed)) => compressed,
Err(_) => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::UnsupportedCompression,
};
}
};
if oi.actual_size < -1 || (oi.actual_size == -1 && !compressed) {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidObjectSize,
};
}
// Match ObjectInfo::get_actual_size: a positive in-memory value is the
// authoritative decoded size. Stale declared/part metadata must not turn
// an otherwise valid object into a false corruption report.
if oi.actual_size > 0 {
return SizeResolution::Known {
logical: oi.actual_size,
physical,
};
}
if oi
.parts
.iter()
.any(|part| part.actual_size < -1 || (part.actual_size < 0 && !compressed))
{
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidPartSize,
};
}
let declared = rustfs_utils::http::get_str(&oi.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE);
let declared = match declared {
Some(value) if value.is_empty() => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidDeclaredSize,
};
}
Some(value) => match value.parse::<i64>() {
Ok(value) if value >= 0 => Some(value),
_ => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidDeclaredSize,
};
}
},
None => None,
};
let logical = match oi.get_actual_size() {
Ok(size) if size == -1 && compressed && declared.is_none() => {
return SizeResolution::Unknown {
physical,
reason: SizeResolutionReason::CompressedSizeUnknown,
};
}
Ok(size) if size >= 0 => size,
Ok(_) | Err(_) => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::SizeOverflowOrMismatch,
};
}
};
if compressed && logical == 0 && physical != 0 && oi.parts.is_empty() && declared.is_none() {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::SizeOverflowOrMismatch,
};
}
SizeResolution::Known { logical, physical }
}
fn resolve_sizes(object_infos: &[ObjectInfo]) -> Vec<SizeResolution> {
object_infos.iter().map(resolve_size).collect()
}
fn lifecycle_rule_has_size_filter(lifecycle: &BucketLifecycleConfiguration, rule_id: &str) -> bool {
let filter_has_size = |filter: &s3s::dto::LifecycleRuleFilter| {
filter.object_size_greater_than.is_some()
|| filter.object_size_less_than.is_some()
|| filter
.and
.as_ref()
.is_some_and(|and| and.object_size_greater_than.is_some() || and.object_size_less_than.is_some())
};
lifecycle
.rules
.iter()
.find(|rule| rule.id.as_deref().unwrap_or_default() == rule_id)
.and_then(|rule| rule.filter.as_ref())
.is_some_and(filter_has_size)
}
fn lifecycle_event_allowed(resolution: &SizeResolution, event: &Event, lifecycle: &BucketLifecycleConfiguration) -> bool {
match resolution {
// Corrupt metadata cannot safely authorize a destructive action, even
// when the evaluator happened to produce a time-only event.
SizeResolution::Corrupt { .. } => false,
// A valid-but-unknown logical size may still execute lifecycle
// actions whose rule is independent of object-size predicates. The
// evaluator has already selected the rule; only that rule's filter
// can make the missing logical value action-critical.
SizeResolution::Unknown { .. } => !lifecycle_rule_has_size_filter(lifecycle, &event.rule_id),
SizeResolution::Known { .. } => true,
}
}
/// A successful newer-noncurrent batch consumes both known and unresolved
/// versions from the retained-version alert count. The two accounting paths
/// are separate because only known sizes can contribute byte totals.
fn remaining_versions_after_queued_noncurrent(remaining_versions: usize, known_count: usize, unknown_count: usize) -> usize {
remaining_versions.saturating_sub(known_count.saturating_add(unknown_count))
}
/// How the corrupt-metadata branch records the repair after attempting an
/// MRF intent (backlog#1894 axis A).
#[derive(Debug, PartialEq, Eq)]
@@ -319,34 +573,45 @@ impl ScannerItem {
"Scanner lifecycle evaluation started"
);
let resolved_sizes = resolve_sizes(&object_infos);
if let Some(first) = object_infos.first() {
size_summary.record_reconciliation_scope(
&bounded_reconciliation_field(&first.bucket),
&bounded_reconciliation_field(&first.name),
);
}
for (oi, resolution) in object_infos.iter().zip(resolved_sizes.iter()) {
record_size_resolution(size_summary, oi, resolution);
}
let has_corrupt_size = resolved_sizes
.iter()
.any(|resolution| matches!(resolution, SizeResolution::Corrupt { .. }));
// `versioning_config` is resolved once per object by the caller
// (`get_size`) and handed in; only `prefix_enabled` is consulted here.
let Some(lifecycle) = self.lifecycle.as_ref() else {
let mut cumulative_size = 0;
for oi in object_infos.iter() {
let actual_size = match oi.get_actual_size() {
Ok(size) => size,
Err(_) => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %oi.name,
state = "size_lookup_failed",
"Scanner lifecycle action used fallback size"
);
let Some(lifecycle) = self.lifecycle.clone() else {
let mut cumulative_size: i64 = 0;
for (oi, resolved_size) in object_infos.iter().zip(resolved_sizes.iter()) {
let accounting_size = match resolved_size {
SizeResolution::Known { logical, .. } => *logical,
// A valid compressed legacy sentinel has no logical size,
// but heal and replication still need to run. The
// physical size is only an input to those operations; it
// is not folded into the logical total below.
SizeResolution::Unknown { physical, .. } => {
self.heal_actions(oi, *physical, size_summary).await;
size_summary.actions_accounting_unknown(oi);
continue;
}
SizeResolution::Corrupt { .. } => continue,
};
let size = self.heal_actions(oi, actual_size, size_summary).await;
let size = self.heal_actions(oi, accounting_size, size_summary).await;
size_summary.actions_accounting(oi, size, actual_size);
size_summary.actions_accounting(oi, size, accounting_size);
cumulative_size += size;
cumulative_size = cumulative_size.saturating_add(size);
}
self.alert_excessive_versions(object_infos.len(), cumulative_size);
@@ -400,25 +665,108 @@ impl ScannerItem {
let mut to_delete_objs: Vec<ObjectToDelete> = Vec::new();
let mut noncurrent_events: Vec<Event> = Vec::new();
let mut noncurrent_accounting: Vec<PendingScannerAccounting<'_>> = Vec::new();
let mut noncurrent_unknown: Vec<&ObjectInfo> = Vec::new();
let mut cumulative_size = 0;
let mut remaining_versions = object_infos.len();
'eventLoop: {
for (i, event) in events.iter().enumerate() {
let oi = &object_infos[i];
let actual_size = match oi.get_actual_size() {
Ok(size) => size,
Err(_) => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %oi.name,
state = "size_lookup_failed",
"Scanner lifecycle action used fallback size"
);
0
let known_size = resolved_sizes[i].known_size();
if has_corrupt_size
&& matches!(
event.action,
IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction
)
{
// An all-version delete would also remove a corrupt
// sibling that could not be reconciled safely.
continue;
}
if !lifecycle_event_allowed(&resolved_sizes[i], event, &lifecycle) {
// An unknown logical size must not make an otherwise
// non-destructive scan disappear from heal/physical-tier
// accounting. Size-filtered or deferred events remain
// pending, so retain the version-only physical counters.
if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
self.heal_actions(oi, *physical, size_summary).await;
size_summary.actions_accounting_unknown(oi);
}
continue;
}
let actual_size = match known_size {
Some(size) => size,
None => {
match event.action {
IlmAction::DeleteAction
| IlmAction::DeleteRestoredAction
| IlmAction::DeleteRestoredVersionAction
| IlmAction::DeleteAllVersionsAction
| IlmAction::DelMarkerDeleteAllVersionsAction => {
let done_ilm = Metrics::time_ilm(event.action);
let trace_started_at = trace_start_instant();
let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await;
emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at);
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
done_ilm(1)();
if event.action == IlmAction::DeleteAllVersionsAction
|| event.action == IlmAction::DelMarkerDeleteAllVersionsAction
{
remaining_versions = 0;
}
} else if matches!(
event.action,
IlmAction::DeleteAction
| IlmAction::DeleteRestoredAction
| IlmAction::DeleteRestoredVersionAction
) {
size_summary.actions_accounting_unknown(oi);
} else {
size_summary.actions_accounting_unknown(oi);
for (j, retained) in object_infos.iter().enumerate().skip(i + 1) {
match &resolved_sizes[j] {
SizeResolution::Known { logical, .. } => PendingScannerAccounting {
object: retained,
retained_size: *logical,
expired_size: 0,
}
.apply(size_summary, &mut cumulative_size, false),
SizeResolution::Unknown { .. } => {
size_summary.actions_accounting_unknown(retained);
}
SizeResolution::Corrupt { .. } => {}
}
}
}
}
IlmAction::DeleteVersionAction => {
if let Some(opt) = object_opts.get(i) {
to_delete_objs.push(ObjectToDelete {
object_name: opt.name.clone(),
version_id: opt.version_id,
..Default::default()
});
noncurrent_events.push(event.clone());
noncurrent_unknown.push(oi);
}
}
IlmAction::TransitionAction | IlmAction::TransitionVersionAction => {
let trace_started_at = trace_start_instant();
let queued = apply_transition_rule(event, &LcEventSrc::Scanner, oi).await;
emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at);
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
let done_ilm = Metrics::time_ilm(event.action);
done_ilm(1)();
}
size_summary.actions_accounting_unknown(oi);
}
IlmAction::NoneAction | IlmAction::ActionCount => {
if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
self.heal_actions(oi, *physical, size_summary).await;
}
size_summary.actions_accounting_unknown(oi);
}
}
continue;
}
};
@@ -446,36 +794,24 @@ impl ScannerItem {
done_ilm(1)();
remaining_versions = 0;
} else {
PendingScannerAccounting {
object: oi,
retained_size: actual_size,
expired_size: 0,
}
.apply(size_summary, &mut cumulative_size, false);
for retained in object_infos.iter().skip(i + 1) {
let retained_size = match retained.get_actual_size() {
Ok(size) => size,
Err(_) => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %retained.name,
state = "size_lookup_failed",
"Scanner lifecycle action used fallback size"
);
0
}
};
if let Some(actual_size) = known_size {
PendingScannerAccounting {
object: retained,
retained_size,
object: oi,
retained_size: actual_size,
expired_size: 0,
}
.apply(size_summary, &mut cumulative_size, false);
}
for (j, retained) in object_infos.iter().enumerate().skip(i + 1) {
if let Some(retained_size) = resolved_sizes[j].known_size() {
PendingScannerAccounting {
object: retained,
retained_size,
expired_size: 0,
}
.apply(size_summary, &mut cumulative_size, false);
}
}
}
break 'eventLoop;
}
@@ -511,11 +847,13 @@ impl ScannerItem {
version_id: opt.version_id,
..Default::default()
});
noncurrent_accounting.push(PendingScannerAccounting {
object: oi,
retained_size: actual_size,
expired_size: 0,
});
if let Some(actual_size) = known_size {
noncurrent_accounting.push(PendingScannerAccounting {
object: oi,
retained_size: actual_size,
expired_size: 0,
});
}
account_now = false;
}
noncurrent_events.push(event.clone());
@@ -548,7 +886,7 @@ impl ScannerItem {
if account_now {
size_summary.actions_accounting(oi, size, actual_size);
cumulative_size += size;
cumulative_size = cumulative_size.saturating_add(size);
}
}
}
@@ -576,11 +914,20 @@ impl ScannerItem {
}
if record_scanner_ilm_action_if_queued(global_metrics(), action, count, queued) {
done_ilm(count)();
remaining_versions = remaining_versions.saturating_sub(noncurrent_accounting.len());
remaining_versions = remaining_versions_after_queued_noncurrent(
remaining_versions,
noncurrent_accounting.len(),
noncurrent_unknown.len(),
);
}
for pending in noncurrent_accounting {
pending.apply(size_summary, &mut cumulative_size, queued);
}
if !queued {
for object in noncurrent_unknown {
size_summary.actions_accounting_unknown(object);
}
}
}
self.alert_excessive_versions(remaining_versions, cumulative_size);
}
@@ -929,4 +1276,361 @@ mod tests {
assert_eq!(item.object_name, "object");
assert_eq!(item.object_path(), "object");
}
#[test]
fn size_resolution_rejects_negative_overflow_and_unknown_compression() {
let compressed = |actual_size: i64, declared: Option<&str>| {
let mut user_defined = HashMap::new();
rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
if let Some(declared) = declared {
rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, declared.to_string());
}
ObjectInfo {
size: 12,
actual_size,
user_defined: Arc::new(user_defined),
..Default::default()
}
};
let normal = ObjectInfo {
size: 12,
actual_size: 10,
..Default::default()
};
assert_eq!(
resolve_size(&normal),
SizeResolution::Known {
logical: 10,
physical: 12
}
);
let stale_declared_metadata = ObjectInfo {
size: 12,
actual_size: 10,
user_defined: Arc::new(HashMap::from([("x-rustfs-internal-actual-size".to_string(), "not-a-size".to_string())])),
parts: Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
actual_size: -2,
..Default::default()
}]),
..Default::default()
};
assert_eq!(
resolve_size(&stale_declared_metadata),
SizeResolution::Known {
logical: 10,
physical: 12
}
);
assert_eq!(
resolve_size(&compressed(0, Some("9"))),
SizeResolution::Known {
logical: 9,
physical: 12
}
);
assert_eq!(
resolve_size(&compressed(-1, None)),
SizeResolution::Unknown {
physical: 12,
reason: SizeResolutionReason::CompressedSizeUnknown,
}
);
assert!(matches!(
resolve_size(&compressed(0, Some("not-a-size"))),
SizeResolution::Corrupt {
reason: SizeResolutionReason::InvalidDeclaredSize,
..
}
));
assert!(matches!(
resolve_size(&ObjectInfo {
size: 12,
actual_size: -2,
..Default::default()
}),
SizeResolution::Corrupt { .. }
));
assert!(matches!(resolve_size(&compressed(0, Some("-1"))), SizeResolution::Corrupt { .. }));
assert!(matches!(resolve_size(&compressed(0, Some(""))), SizeResolution::Corrupt { .. }));
let unsupported = {
let mut object = compressed(0, None);
let mut metadata = (*object.user_defined).clone();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "unsupported".to_string());
object.user_defined = Arc::new(metadata);
object
};
assert!(matches!(resolve_size(&unsupported), SizeResolution::Corrupt { .. }));
let invalid_part = {
let mut object = compressed(0, None);
object.parts = Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
size: 12,
actual_size: -2,
..Default::default()
}]);
object
};
assert!(matches!(resolve_size(&invalid_part), SizeResolution::Corrupt { .. }));
let overflow = {
let mut object = compressed(0, None);
object.parts = Arc::new(vec![
rustfs_filemeta::ObjectPartInfo {
size: 1,
actual_size: i64::MAX,
..Default::default()
},
rustfs_filemeta::ObjectPartInfo {
size: 1,
actual_size: 1,
..Default::default()
},
]);
object
};
assert!(matches!(resolve_size(&overflow), SizeResolution::Corrupt { .. }));
let mismatch = compressed(0, None);
assert!(matches!(resolve_size(&mismatch), SizeResolution::Corrupt { .. }));
assert_eq!(
resolve_size(&ObjectInfo {
size: 0,
actual_size: 0,
..Default::default()
}),
SizeResolution::Known { logical: 0, physical: 0 }
);
}
#[test]
fn size_resolution_records_and_replays_one_identity() {
let version_id = uuid::Uuid::new_v4();
let generation = uuid::Uuid::new_v4();
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "not-a-number".to_string());
let corrupt = ObjectInfo {
bucket: "bucket".to_string(),
name: "object".to_string(),
size: 12,
version_id: Some(version_id),
data_dir: Some(generation),
user_defined: Arc::new(metadata),
..Default::default()
};
let mut summary = SizeSummary::default();
let resolution = resolve_size(&corrupt);
record_size_resolution(&mut summary, &corrupt, &resolution);
record_size_resolution(&mut summary, &corrupt, &resolution);
assert_eq!(summary.size_reconciliation.len(), 1);
assert_eq!(summary.size_reconciliation[0].reason, "invalid_declared_size");
assert_eq!(summary.size_reconciliation[0].physical_size, Some(12));
let known = ObjectInfo {
actual_size: 12,
user_defined: Arc::new(HashMap::new()),
..corrupt.clone()
};
record_size_resolution(&mut summary, &known, &resolve_size(&known));
summary.record_reconciliation_scope(&known.bucket, &known.name);
assert_eq!(summary.reconciliation_scopes.len(), 1);
assert_eq!(summary.reconciliation_scopes[0].bucket, "bucket");
}
#[test]
fn malformed_size_has_same_ilm_accounting() {
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "invalid".to_string());
let object = ObjectInfo {
bucket: "bucket".to_string(),
name: "object".to_string(),
size: 12,
user_defined: Arc::new(metadata),
..Default::default()
};
let resolution = resolve_size(&object);
let mut without_ilm = SizeSummary::default();
let mut with_ilm = SizeSummary::default();
record_size_resolution(&mut without_ilm, &object, &resolution);
record_size_resolution(&mut with_ilm, &object, &resolution);
assert_eq!(without_ilm.size_reconciliation, with_ilm.size_reconciliation);
assert_eq!(without_ilm.total_size, 0);
assert_eq!(with_ilm.total_size, 0);
assert!(without_ilm.tier_stats.is_empty());
assert!(with_ilm.tier_stats.is_empty());
}
#[test]
fn size_resolution_parses_once_per_version() {
let objects = vec![
ObjectInfo {
bucket: "bucket".to_string(),
name: "one".to_string(),
size: 1,
actual_size: 1,
..Default::default()
},
ObjectInfo {
bucket: "bucket".to_string(),
name: "two".to_string(),
size: 2,
actual_size: -2,
..Default::default()
},
];
let resolutions = resolve_sizes(&objects);
assert_eq!(resolutions.len(), objects.len());
assert!(matches!(resolutions[0], SizeResolution::Known { logical: 1, .. }));
assert!(matches!(resolutions[1], SizeResolution::Corrupt { .. }));
}
#[test]
fn queued_unknown_noncurrent_versions_are_removed_from_alert_count() {
assert_eq!(remaining_versions_after_queued_noncurrent(3, 1, 2), 0);
assert_eq!(remaining_versions_after_queued_noncurrent(7, 2, 1), 4);
assert_eq!(remaining_versions_after_queued_noncurrent(usize::MAX, usize::MAX, usize::MAX), 0);
}
#[test]
fn malformed_size_blocks_size_dependent_transition_but_allows_time_only_expiry() {
let size_filtered = BucketLifecycleConfiguration {
rules: vec![s3s::dto::LifecycleRule {
status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
expiration: None,
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
id: Some("size".to_string()),
filter: Some(s3s::dto::LifecycleRuleFilter {
object_size_greater_than: Some(1),
..Default::default()
}),
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
transitions: None,
}],
..Default::default()
};
let unknown = SizeResolution::Unknown {
physical: 12,
reason: SizeResolutionReason::CompressedSizeUnknown,
};
let size_event = Event {
action: IlmAction::DeleteAction,
rule_id: "size".to_string(),
..Default::default()
};
assert!(!lifecycle_event_allowed(&unknown, &size_event, &size_filtered));
assert!(!lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::TransitionAction,
rule_id: "size".to_string(),
..Default::default()
},
&size_filtered
));
let mixed_filters = BucketLifecycleConfiguration {
rules: vec![
size_filtered.rules[0].clone(),
s3s::dto::LifecycleRule {
status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
expiration: None,
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
id: Some("time".to_string()),
filter: None,
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
transitions: None,
},
],
..Default::default()
};
assert!(lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::DeleteAction,
rule_id: "time".to_string(),
..Default::default()
},
&mixed_filters
));
assert!(lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::TransitionAction,
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
assert!(!lifecycle_event_allowed(
&SizeResolution::Corrupt {
physical: 12,
reason: SizeResolutionReason::InvalidDeclaredSize,
},
&Event {
action: IlmAction::DeleteAction,
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
assert!(lifecycle_event_allowed(
&SizeResolution::Known {
logical: 10,
physical: 12,
},
&Event {
action: IlmAction::DeleteAllVersionsAction,
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
assert!(lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::DeleteAction,
rule_id: "time-only".to_string(),
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
}
#[tokio::test]
async fn long_object_size_reconciliation_scope_uses_bounded_identity() {
let object_name = "o".repeat(600);
let mut item = scanner_item_with_prefix("");
item.object_name = object_name.clone();
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
let object = ObjectInfo {
bucket: item.bucket.clone(),
name: object_name.clone(),
size: 12,
actual_size: -1,
version_id: Some(uuid::Uuid::new_v4()),
user_defined: Arc::new(metadata),
..Default::default()
};
let mut summary = SizeSummary::default();
item.apply_actions(vec![object], None, VersioningConfiguration::default(), &mut summary)
.await;
let bounded_bucket = bounded_reconciliation_field(&item.bucket);
let bounded_object = bounded_reconciliation_field(&object_name);
assert_eq!(summary.reconciliation_scopes[0].bucket, bounded_bucket);
assert_eq!(summary.reconciliation_scopes[0].object, bounded_object);
assert_eq!(summary.size_reconciliation[0].object, bounded_object);
assert_eq!(summary.versions, 1);
assert_eq!(summary.total_size, 0);
}
}
@@ -388,6 +388,66 @@ async fn test_record_failed_ttl_zero_noop() {
assert!(!scanner.should_skip_failed("path2"));
}
#[tokio::test]
async fn malformed_size_reconciliation_replays_after_restart() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
let entry = SizeReconciliationEntry {
key: "1:b|6:object|0:|0:".to_string(),
bucket: "b".to_string(),
object: "object".to_string(),
reason: "invalid_declared_size".to_string(),
physical_size: Some(12),
..Default::default()
};
let mut summary = SizeSummary::default();
summary.record_size_reconciliation(entry.clone());
summary.record_reconciliation_scope("b", "object");
scanner.apply_size_reconciliation(&summary);
scanner.apply_size_reconciliation(&summary);
assert_eq!(scanner.new_cache.info.size_reconciliation.len(), 1);
assert_eq!(scanner.update_cache.info.size_reconciliation.len(), 1);
assert_eq!(scanner.new_cache.info.size_reconciliation[&entry.key].attempts, 2);
let encoded = rmp_serde::to_vec_named(&scanner.new_cache.info).expect("size ledger should encode");
let decoded: crate::data_usage_define::DataUsageCacheInfo =
rmp_serde::from_slice(&encoded).expect("size ledger should decode");
assert_eq!(decoded.size_reconciliation.len(), 1);
assert_eq!(decoded.size_reconciliation[&entry.key].reason, "invalid_declared_size");
let mut resolved = SizeSummary::default();
resolved.record_reconciliation_scope("b", "object");
scanner.apply_size_reconciliation(&resolved);
assert!(scanner.new_cache.info.size_reconciliation.is_empty());
assert!(scanner.update_cache.info.size_reconciliation.is_empty());
}
#[tokio::test]
async fn malformed_size_reconciliation_clears_bounded_long_object_scope() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
let long_object = "o".repeat(600);
let bounded_object = item_actions::bounded_reconciliation_field(&long_object);
let entry = SizeReconciliationEntry {
key: "long-object-key".to_string(),
bucket: "b".to_string(),
object: bounded_object,
reason: "invalid_declared_size".to_string(),
..Default::default()
};
let mut summary = SizeSummary::default();
summary.record_size_reconciliation(entry);
scanner.apply_size_reconciliation(&summary);
assert_eq!(scanner.new_cache.info.size_reconciliation.len(), 1);
let mut resolved = SizeSummary::default();
resolved.record_reconciliation_scope("b", &long_object);
scanner.apply_size_reconciliation(&resolved);
assert!(scanner.new_cache.info.size_reconciliation.is_empty());
}
#[test]
fn test_classify_get_size_failure_marks_metadata_heal_object_path() {
let temp_dir = std::env::temp_dir();
+7 -49
View File
@@ -73,8 +73,6 @@ enum TargetUpdateOp {
/// Connection group: credentials plus endpoint, target bucket, and TLS settings.
Credentials,
Sync,
/// Per-target read-proxy opt-out (`disableProxy`).
Proxy,
Bandwidth,
Path,
}
@@ -83,13 +81,12 @@ fn parse_remote_target_update_ops(queries: &HashMap<String, String>) -> S3Result
const SUPPORTED_OPS: &[(&str, TargetUpdateOp)] = &[
("creds", TargetUpdateOp::Credentials),
("sync", TargetUpdateOp::Sync),
("proxy", TargetUpdateOp::Proxy),
("bandwidth", TargetUpdateOp::Bandwidth),
("path", TargetUpdateOp::Path),
];
// Present in the MinIO wire contract, but they drive target fields this
// version rejects as unsupported — fail loudly instead of silently ignoring.
const UNSUPPORTED_OPS: &[&str] = &["healthcheck", "edge", "edgeSyncBeforeExpiry"];
const UNSUPPORTED_OPS: &[&str] = &["proxy", "healthcheck", "edge", "edgeSyncBeforeExpiry"];
for key in UNSUPPORTED_OPS {
if queries.get(*key).is_some_and(|value| value == "true") {
@@ -315,10 +312,11 @@ impl RemoteTargetRequest {
));
}
for (unsupported, configured) in REMOTE_TARGET_UNSUPPORTED_FIELDS
.iter()
.copied()
.zip([self.edge, self.edge_sync_before_expiry])
for (unsupported, configured) in
REMOTE_TARGET_UNSUPPORTED_FIELDS
.iter()
.copied()
.zip([self.disable_proxy, self.edge, self.edge_sync_before_expiry])
{
if configured {
return Err(s3_error!(
@@ -704,7 +702,6 @@ impl Operation for SetRemoteTargetHandler {
target.deployment_id = remote_target.deployment_id.clone();
}
TargetUpdateOp::Sync => target.replication_sync = remote_target.replication_sync,
TargetUpdateOp::Proxy => target.disable_proxy = remote_target.disable_proxy,
TargetUpdateOp::Bandwidth => target.bandwidth_limit = remote_target.bandwidth_limit,
TargetUpdateOp::Path => target.path = remote_target.path.clone(),
}
@@ -1523,7 +1520,6 @@ mod tests {
("update", "true"),
("creds", "true"),
("sync", "true"),
("proxy", "true"),
("bandwidth", "true"),
("path", "true"),
]))
@@ -1533,7 +1529,6 @@ mod tests {
vec![
TargetUpdateOp::Credentials,
TargetUpdateOp::Sync,
TargetUpdateOp::Proxy,
TargetUpdateOp::Bandwidth,
TargetUpdateOp::Path
]
@@ -2075,6 +2070,7 @@ mod tests {
("credentials.session_token", serde_json::json!("session-token")),
("credentials.expiration", serde_json::json!("2026-01-01T00:00:00Z")),
("api", serde_json::json!("s3v2")),
("disableProxy", serde_json::json!(true)),
("edge", serde_json::json!(true)),
("edgeSyncBeforeExpiry", serde_json::json!(true)),
] {
@@ -2304,44 +2300,6 @@ mod tests {
assert!(!REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"healthCheckDuration"));
}
#[test]
fn remote_target_disable_proxy_is_declared_writable_edge_stays_unsupported() {
assert!(REMOTE_TARGET_WRITABLE_FIELDS.contains(&"disableProxy"));
assert!(!REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"disableProxy"));
// edge sync has no implementation behind it — it must stay rejected.
assert!(REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"edge"));
assert!(REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(&"edgeSyncBeforeExpiry"));
}
#[test]
fn remote_target_create_accepts_disable_proxy() {
let mut request = valid_remote_target_request();
request["disableProxy"] = serde_json::json!(true);
let target = serde_json::from_value::<RemoteTargetRequest>(request)
.expect("request should deserialize")
.into_bucket_target()
.expect("disableProxy is a supported per-target read-proxy opt-out");
assert!(target.disable_proxy);
}
#[test]
fn update_body_with_proxy_op_toggles_disable_proxy_without_credentials() {
// Mirrors the other partial-update groups: a proxy-only update body may
// omit the connection fields entirely.
let body = serde_json::json!({
"arn": "arn:rustfs:replication:us-east-1:dep:target",
"type": "replication",
"disableProxy": true
});
let request: RemoteTargetRequest = serde_json::from_value(body).expect("partial update body should deserialize");
let target = request
.into_update_bucket_target(&[TargetUpdateOp::Proxy])
.expect("proxy-only update must not require credentials");
assert!(target.disable_proxy);
}
#[test]
fn remote_target_capability_fields_do_not_overlap() {
for field in REMOTE_TARGET_UNSUPPORTED_FIELDS {
+4 -21
View File
@@ -1262,9 +1262,7 @@ mod tests {
assert_eq!(response.summary.manual_transition_jobs.state, CapabilityState::Supported);
assert_eq!(response.replication.contract_version, 1);
assert_eq!(response.replication.bucket_replication.contract_version, 1);
// v2: disableProxy moved from unsupported to writable (per-target
// read-proxy opt-out reached the admin API).
assert_eq!(response.replication.remote_targets.contract_version, 2);
assert_eq!(response.replication.remote_targets.contract_version, 1);
assert_eq!(response.replication.bucket_replication.status.state, CapabilityState::Supported);
assert_eq!(response.replication.remote_targets.status.state, CapabilityState::Supported);
assert_eq!(
@@ -1295,15 +1293,7 @@ mod tests {
.remote_targets
.fields
.iter()
.any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Supported)
);
assert!(
response
.replication
.remote_targets
.fields
.iter()
.any(|field| field.name == "edge" && field.state == super::ReplicationFieldState::Unsupported)
.any(|field| field.name == "disableProxy" && field.state == super::ReplicationFieldState::Unsupported)
);
assert!(
response
@@ -1374,7 +1364,7 @@ mod tests {
assert_eq!(value["summary"]["manual_transition_jobs"]["state"], "supported");
assert_eq!(value["replication"]["contract_version"], 1);
assert_eq!(value["replication"]["bucket_replication"]["contract_version"], 1);
assert_eq!(value["replication"]["remote_targets"]["contract_version"], 2);
assert_eq!(value["replication"]["remote_targets"]["contract_version"], 1);
assert_eq!(value["replication"]["bucket_replication"]["status"]["state"], "supported");
assert_eq!(value["replication"]["remote_targets"]["status"]["state"], "supported");
assert_eq!(
@@ -1393,14 +1383,7 @@ mod tests {
.as_array()
.expect("remote target fields should be an array")
.iter()
.any(|field| field["name"] == "disableProxy" && field["state"] == "supported")
);
assert!(
value["replication"]["remote_targets"]["fields"]
.as_array()
.expect("remote target fields should be an array")
.iter()
.any(|field| field["name"] == "edge" && field["state"] == "unsupported")
.any(|field| field["name"] == "disableProxy" && field["state"] == "unsupported")
);
assert!(
value["replication"]["remote_targets"]["fields"]