mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-22 20:36:38 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 23b85b792e |
@@ -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());
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
|
||||
@@ -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", {
|
||||
|
||||
@@ -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(())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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}"),
|
||||
|
||||
@@ -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(())
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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"]
|
||||
|
||||
Reference in New Issue
Block a user