mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-23 04:39:04 +00:00
Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 16fc3ff0e7 | |||
| 6dbff1c036 | |||
| 854b6dfb0c | |||
| 893a4a11c1 | |||
| 2086ade967 | |||
| 23b85b792e |
@@ -307,6 +307,38 @@ pub struct DiskUsageStatus {
|
|||||||
pub snapshot_exists: bool,
|
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
|
/// Size summary for a single object or group of objects
|
||||||
#[derive(Debug, Default, Clone)]
|
#[derive(Debug, Default, Clone)]
|
||||||
pub struct SizeSummary {
|
pub struct SizeSummary {
|
||||||
@@ -336,6 +368,16 @@ pub struct SizeSummary {
|
|||||||
pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
|
pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
|
||||||
/// Per-tier accounting, keyed by storage class or remote tier name
|
/// Per-tier accounting, keyed by storage class or remote tier name
|
||||||
pub tier_stats: HashMap<String, TierStats>,
|
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
|
/// Replication target size summary
|
||||||
@@ -833,7 +875,8 @@ impl DataUsageEntry {
|
|||||||
///
|
///
|
||||||
/// The canonical wire format is written by the hand-written map-encoded
|
/// The canonical wire format is written by the hand-written map-encoded
|
||||||
/// `Serialize` on the scanner-side `DataUsageCacheInfo`
|
/// `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
|
/// This type decodes only the shared subset and is deliberately not
|
||||||
/// `Serialize`: a derived (array) encoding of this 6-field subset would
|
/// `Serialize`: a derived (array) encoding of this 6-field subset would
|
||||||
/// corrupt the cache for scanner readers, so no write path may exist here.
|
/// corrupt the cache for scanner readers, so no write path may exist here.
|
||||||
@@ -1777,6 +1820,51 @@ impl SizeSummary {
|
|||||||
entry.pending_count = entry.pending_count.saturating_add(stats.pending_count);
|
entry.pending_count = entry.pending_count.saturating_add(stats.pending_count);
|
||||||
entry.failed_count = entry.failed_count.saturating_add(stats.failed_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(),
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -54,6 +54,7 @@ const ERR_LIFECYCLE_INVALID_EXPIRED_OBJECT_ALL_VERSIONS: &str =
|
|||||||
"Days must be a positive integer and Date must not be specified inside Expiration with ExpiredObjectAllVersions";
|
"Days must be a positive integer and Date must not be specified inside Expiration with ExpiredObjectAllVersions";
|
||||||
const ERR_LIFECYCLE_INVALID_DEL_MARKER_EXPIRATION_DAYS: &str = "Days must be a positive integer with DelMarkerExpiration";
|
const ERR_LIFECYCLE_INVALID_DEL_MARKER_EXPIRATION_DAYS: &str = "Days must be a positive integer with DelMarkerExpiration";
|
||||||
const ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG: &str = "Rule ID must be at most 255 characters";
|
const ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG: &str = "Rule ID must be at most 255 characters";
|
||||||
|
const ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY: &str = "Rule ID must not be empty";
|
||||||
const ERR_LIFECYCLE_INVALID_RULE_STATUS: &str = "Rule status must be either Enabled or Disabled";
|
const ERR_LIFECYCLE_INVALID_RULE_STATUS: &str = "Rule status must be either Enabled or Disabled";
|
||||||
const ERR_LIFECYCLE_DEL_MARKER_WITH_TAGS: &str = "Rule with DelMarkerExpiration cannot have tags based filtering";
|
const ERR_LIFECYCLE_DEL_MARKER_WITH_TAGS: &str = "Rule with DelMarkerExpiration cannot have tags based filtering";
|
||||||
const ERR_LIFECYCLE_EXPIRED_OBJECT_DELETE_MARKER_WITH_TAGS: &str =
|
const ERR_LIFECYCLE_EXPIRED_OBJECT_DELETE_MARKER_WITH_TAGS: &str =
|
||||||
@@ -402,10 +403,13 @@ impl Lifecycle for BucketLifecycleConfiguration {
|
|||||||
NoncurrentVersionTransitionOps::validate(transition)?;
|
NoncurrentVersionTransitionOps::validate(transition)?;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if let Some(id) = &r.id
|
if let Some(id) = &r.id {
|
||||||
&& id.len() > 255
|
if id.is_empty() {
|
||||||
{
|
return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY));
|
||||||
return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG));
|
}
|
||||||
|
if id.len() > 255 {
|
||||||
|
return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
r.validate()?;
|
r.validate()?;
|
||||||
if let Some(object_lock_enabled) = lr.object_lock_enabled.as_ref()
|
if let Some(object_lock_enabled) = lr.object_lock_enabled.as_ref()
|
||||||
@@ -3730,6 +3734,31 @@ mod tests {
|
|||||||
.expect("empty prefix with filter should be valid");
|
.expect("empty prefix with filter should be valid");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn validate_rejects_empty_rule_id() {
|
||||||
|
let lc = BucketLifecycleConfiguration {
|
||||||
|
expiry_updated_at: None,
|
||||||
|
rules: vec![LifecycleRule {
|
||||||
|
status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
|
||||||
|
expiration: Some(LifecycleExpiration {
|
||||||
|
days: Some(30),
|
||||||
|
..Default::default()
|
||||||
|
}),
|
||||||
|
abort_incomplete_multipart_upload: None,
|
||||||
|
del_marker_expiration: None,
|
||||||
|
filter: None,
|
||||||
|
id: Some(String::new()),
|
||||||
|
noncurrent_version_expiration: None,
|
||||||
|
noncurrent_version_transitions: None,
|
||||||
|
prefix: None,
|
||||||
|
transitions: None,
|
||||||
|
}],
|
||||||
|
};
|
||||||
|
|
||||||
|
let error = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
|
||||||
|
assert_eq!(error.to_string(), ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY);
|
||||||
|
}
|
||||||
|
|
||||||
// --- TASK-004 tests: ExpiredObjectAllVersions ---
|
// --- TASK-004 tests: ExpiredObjectAllVersions ---
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
|
|||||||
@@ -29,7 +29,8 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
|
|||||||
pub use rustfs_data_usage::{
|
pub use rustfs_data_usage::{
|
||||||
AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
|
AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
|
||||||
DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
|
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 rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
|
||||||
use tokio::time::{Duration, Instant, sleep, timeout};
|
use tokio::time::{Duration, Instant, sleep, timeout};
|
||||||
@@ -192,6 +193,10 @@ const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
|
|||||||
pub trait ScannerSizeSummaryExt {
|
pub trait ScannerSizeSummaryExt {
|
||||||
/// Fold one object's contribution into the summary, including its tier.
|
/// Fold one object's contribution into the summary, including its tier.
|
||||||
fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64);
|
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 {
|
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 =====
|
// ===== Cache-related data structures =====
|
||||||
@@ -344,6 +377,10 @@ pub struct DataUsageCacheInfo {
|
|||||||
pub scan_plan_digest: Option<DataUsageScanPlanDigest>,
|
pub scan_plan_digest: Option<DataUsageScanPlanDigest>,
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub cache_key_format: u16,
|
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 {
|
impl Serialize for DataUsageCacheInfo {
|
||||||
@@ -353,7 +390,8 @@ impl Serialize for DataUsageCacheInfo {
|
|||||||
{
|
{
|
||||||
// Keep this metadata map-encoded so older readers can ignore fields
|
// Keep this metadata map-encoded so older readers can ignore fields
|
||||||
// appended by newer scanner versions during rolling upgrades.
|
// 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("name", &self.name)?;
|
||||||
state.serialize_entry("next_cycle", &self.next_cycle)?;
|
state.serialize_entry("next_cycle", &self.next_cycle)?;
|
||||||
state.serialize_entry("leader_epoch", &self.leader_epoch)?;
|
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("snapshot_complete", &self.snapshot_complete)?;
|
||||||
state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?;
|
state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?;
|
||||||
state.serialize_entry("cache_key_format", &self.cache_key_format)?;
|
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()
|
state.end()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -428,14 +469,18 @@ impl DataUsageCache {
|
|||||||
self.checked_flatten(name).is_some()
|
self.checked_flatten(name).is_some()
|
||||||
});
|
});
|
||||||
if !reusable {
|
if !reusable {
|
||||||
let pending_heals = if self.info.name == name {
|
let (pending_heals, size_reconciliation) = if self.info.name == name {
|
||||||
std::mem::take(&mut self.info.pending_heals)
|
(
|
||||||
|
std::mem::take(&mut self.info.pending_heals),
|
||||||
|
std::mem::take(&mut self.info.size_reconciliation),
|
||||||
|
)
|
||||||
} else {
|
} else {
|
||||||
Vec::new()
|
(Vec::new(), HashMap::new())
|
||||||
};
|
};
|
||||||
*self = Self::default();
|
*self = Self::default();
|
||||||
self.info.name = name.to_string();
|
self.info.name = name.to_string();
|
||||||
self.info.pending_heals = pending_heals;
|
self.info.pending_heals = pending_heals;
|
||||||
|
self.info.size_reconciliation = size_reconciliation;
|
||||||
}
|
}
|
||||||
|
|
||||||
self.info.next_cycle = next_cycle;
|
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]
|
#[test]
|
||||||
fn test_data_usage_entry_merge_sums_failed_objects() {
|
fn test_data_usage_entry_merge_sums_failed_objects() {
|
||||||
let mut left = DataUsageEntry {
|
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),
|
scan_plan_digest: Some(TEST_PLAN_DIGEST),
|
||||||
cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
|
cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
|
||||||
pending_heals: vec![pending_heal.clone()],
|
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()
|
||||||
},
|
},
|
||||||
..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!(outcome, DataUsageCachePrepareOutcome::Reused);
|
||||||
assert_eq!(cache.info.pending_heals, vec![pending_heal]);
|
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.cache.is_empty());
|
||||||
assert!(!cache.info.snapshot_complete);
|
assert!(!cache.info.snapshot_complete);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -20,8 +20,9 @@ use std::time::{Duration, Instant, SystemTime};
|
|||||||
|
|
||||||
use crate::ReplTargetSizeSummary;
|
use crate::ReplTargetSizeSummary;
|
||||||
use crate::data_usage_define::{
|
use crate::data_usage_define::{
|
||||||
DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageScanCheckpoint,
|
DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageCacheInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap,
|
||||||
DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt, SizeSummary, hash_path,
|
DataUsageScanCheckpoint, DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt,
|
||||||
|
SizeReconciliationEntry, SizeSummary, hash_path,
|
||||||
};
|
};
|
||||||
use crate::error::ScannerError;
|
use crate::error::ScannerError;
|
||||||
use crate::runtime_config::{
|
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_PRUNE_TOTAL: &str = "rustfs_scanner_pending_heal_prune_total";
|
||||||
const METRIC_SCANNER_PENDING_HEAL_MALFORMED_TOTAL: &str = "rustfs_scanner_pending_heal_malformed_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_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) --
|
// --- 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) {
|
fn apply(self, size_summary: &mut SizeSummary, cumulative_size: &mut i64, queued: bool) {
|
||||||
let size = if queued { self.expired_size } else { self.retained_size };
|
let size = if queued { self.expired_size } else { self.retained_size };
|
||||||
size_summary.actions_accounting(self.object, size, self.retained_size);
|
size_summary.actions_accounting(self.object, size, self.retained_size);
|
||||||
*cumulative_size += size;
|
*cumulative_size = cumulative_size.saturating_add(size);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -671,10 +675,65 @@ pub struct FolderScanner {
|
|||||||
skip_heal: Arc<std::sync::atomic::AtomicBool>,
|
skip_heal: Arc<std::sync::atomic::AtomicBool>,
|
||||||
local_disk: Arc<Disk>,
|
local_disk: Arc<Disk>,
|
||||||
pending_heals_changed: bool,
|
pending_heals_changed: bool,
|
||||||
|
pending_size_reconciliation_keys: HashSet<String>,
|
||||||
|
pending_size_reconciliation_scopes: HashSet<String>,
|
||||||
|
pending_size_reconciliation_truncated: bool,
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
|
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 size_reconciliation_scope_key(bucket: &str, object: &str) -> String {
|
||||||
|
format!("{}:{}|{}:{}", bucket.len(), bucket, object.len(), object)
|
||||||
|
}
|
||||||
|
|
||||||
|
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 {
|
impl FolderScanner {
|
||||||
fn now_secs() -> u64 {
|
fn now_secs() -> u64 {
|
||||||
SystemTime::now()
|
SystemTime::now()
|
||||||
@@ -748,6 +807,60 @@ 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();
|
||||||
|
self.pending_size_reconciliation_keys
|
||||||
|
.extend(summary.size_reconciliation.iter().map(|entry| entry.key.clone()));
|
||||||
|
self.pending_size_reconciliation_scopes.extend(
|
||||||
|
summary
|
||||||
|
.reconciliation_scopes
|
||||||
|
.iter()
|
||||||
|
.map(|scope| size_reconciliation_scope_key(&scope.bucket, &scope.object)),
|
||||||
|
);
|
||||||
|
self.pending_size_reconciliation_truncated |= summary.size_reconciliation_truncated;
|
||||||
|
|
||||||
|
for info in [&mut self.new_cache.info, &mut self.update_cache.info] {
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn finish_size_reconciliation_batch(&mut self) {
|
||||||
|
let now = Self::now_secs();
|
||||||
|
let current_keys = std::mem::take(&mut self.pending_size_reconciliation_keys);
|
||||||
|
let scopes = std::mem::take(&mut self.pending_size_reconciliation_scopes);
|
||||||
|
let truncated = std::mem::replace(&mut self.pending_size_reconciliation_truncated, false);
|
||||||
|
|
||||||
|
for info in [&mut self.new_cache.info, &mut self.update_cache.info] {
|
||||||
|
if !truncated {
|
||||||
|
info.size_reconciliation.retain(|key, entry| {
|
||||||
|
!scopes.contains(&size_reconciliation_scope_key(&entry.bucket, &entry.object)) || current_keys.contains(key)
|
||||||
|
});
|
||||||
|
}
|
||||||
|
prune_size_reconciliation(info, now);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn record_scan_resume_hint(&mut self, folder: &str) {
|
fn record_scan_resume_hint(&mut self, folder: &str) {
|
||||||
self.new_cache.info.scan_resume_after = Some(folder.to_string());
|
self.new_cache.info.scan_resume_after = Some(folder.to_string());
|
||||||
self.update_cache.info.scan_resume_after = Some(folder.to_string());
|
self.update_cache.info.scan_resume_after = Some(folder.to_string());
|
||||||
@@ -1426,6 +1539,7 @@ impl FolderScanner {
|
|||||||
abandoned_children.remove(&path_join_buf(&[&item.bucket, &item.object_path()]));
|
abandoned_children.remove(&path_join_buf(&[&item.bucket, &item.object_path()]));
|
||||||
|
|
||||||
apply_scanner_size_summary(into, &sz);
|
apply_scanner_size_summary(into, &sz);
|
||||||
|
self.apply_size_reconciliation(&sz);
|
||||||
into.objects += 1;
|
into.objects += 1;
|
||||||
object_count += 1;
|
object_count += 1;
|
||||||
self.budget.record_object_scanned();
|
self.budget.record_object_scanned();
|
||||||
@@ -2105,6 +2219,7 @@ impl FolderScanner {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
self.finish_size_reconciliation_batch();
|
||||||
done_folder();
|
done_folder();
|
||||||
let scanned_objects = u64::try_from(into.objects).unwrap_or(u64::MAX);
|
let scanned_objects = u64::try_from(into.objects).unwrap_or(u64::MAX);
|
||||||
emit_scanner_folder_trace(&self.root, &folder.name, scanned_objects, trace_started_at, "completed");
|
emit_scanner_folder_trace(&self.root, &folder.name, scanned_objects, trace_started_at, "completed");
|
||||||
@@ -2190,10 +2305,17 @@ pub async fn scan_data_folder(
|
|||||||
skip_heal,
|
skip_heal,
|
||||||
local_disk,
|
local_disk,
|
||||||
pending_heals_changed: false,
|
pending_heals_changed: false,
|
||||||
|
pending_size_reconciliation_keys: HashSet::new(),
|
||||||
|
pending_size_reconciliation_scopes: HashSet::new(),
|
||||||
|
pending_size_reconciliation_truncated: false,
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
list_path_raw_options_observer: None,
|
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
|
// Check if context is cancelled
|
||||||
if ctx.is_cancelled() {
|
if ctx.is_cancelled() {
|
||||||
return Err(ScannerError::Other("Operation cancelled".to_string()));
|
return Err(ScannerError::Other("Operation cancelled".to_string()));
|
||||||
@@ -2217,7 +2339,9 @@ pub async fn scan_data_folder(
|
|||||||
new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
|
new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
|
||||||
new_cache.info.last_update = Some(SystemTime::now());
|
new_cache.info.last_update = Some(SystemTime::now());
|
||||||
new_cache.info.next_cycle = cache.info.next_cycle;
|
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;
|
new_cache.info.snapshot_complete = !unresolved_objects;
|
||||||
let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some();
|
let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some();
|
||||||
new_cache.info.scan_resume_after = None;
|
new_cache.info.scan_resume_after = None;
|
||||||
@@ -2245,7 +2369,7 @@ pub async fn scan_data_folder(
|
|||||||
if root_has_progress {
|
if root_has_progress {
|
||||||
new_cache.replace_hashed(&root_hash, &None, &root);
|
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() {
|
if new_cache.root().is_some() {
|
||||||
new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
|
new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -13,6 +13,7 @@
|
|||||||
// limitations under the License.
|
// limitations under the License.
|
||||||
/// Per-object scan actions: ScannerItem, the get-size failure policy, and the heal/ILM admission helpers.
|
/// Per-object scan actions: ScannerItem, the get-size failure policy, and the heal/ILM admission helpers.
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use sha2::{Digest as _, Sha256};
|
||||||
|
|
||||||
/// Cached folder information for scanning
|
/// Cached folder information for scanning
|
||||||
#[derive(Clone, Debug)]
|
#[derive(Clone, Debug)]
|
||||||
@@ -32,6 +33,263 @@ pub(super) enum GetSizeFailureAction {
|
|||||||
HealMetadata { object: String },
|
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| {
|
||||||
|
if rule_id.is_empty() {
|
||||||
|
rule.id.as_deref().is_none_or(str::is_empty)
|
||||||
|
} else {
|
||||||
|
rule.id.as_deref() == Some(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 {
|
||||||
|
// Missing or invalid logical size only defers actions whose selected
|
||||||
|
// rule actually depends on that size. Time/version-only actions retain
|
||||||
|
// their existing semantics, including intrinsic events without a rule ID.
|
||||||
|
SizeResolution::Unknown { .. } | SizeResolution::Corrupt { .. } => {
|
||||||
|
!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
|
/// How the corrupt-metadata branch records the repair after attempting an
|
||||||
/// MRF intent (backlog#1894 axis A).
|
/// MRF intent (backlog#1894 axis A).
|
||||||
#[derive(Debug, PartialEq, Eq)]
|
#[derive(Debug, PartialEq, Eq)]
|
||||||
@@ -319,34 +577,48 @@ impl ScannerItem {
|
|||||||
"Scanner lifecycle evaluation started"
|
"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
|
// `versioning_config` is resolved once per object by the caller
|
||||||
// (`get_size`) and handed in; only `prefix_enabled` is consulted here.
|
// (`get_size`) and handed in; only `prefix_enabled` is consulted here.
|
||||||
|
|
||||||
let Some(lifecycle) = self.lifecycle.as_ref() else {
|
let Some(lifecycle) = self.lifecycle.clone() else {
|
||||||
let mut cumulative_size = 0;
|
let mut cumulative_size: i64 = 0;
|
||||||
for oi in object_infos.iter() {
|
for (oi, resolved_size) in object_infos.iter().zip(resolved_sizes.iter()) {
|
||||||
let actual_size = match oi.get_actual_size() {
|
let accounting_size = match resolved_size {
|
||||||
Ok(size) => size,
|
SizeResolution::Known { logical, .. } => *logical,
|
||||||
Err(_) => {
|
// A valid compressed legacy sentinel has no logical size,
|
||||||
warn!(
|
// but heal and replication still need to run. The
|
||||||
target: "rustfs::scanner::folder",
|
// physical size is only an input to those operations; it
|
||||||
event = EVENT_SCANNER_LIFECYCLE_ACTION,
|
// is not folded into the logical total below.
|
||||||
component = LOG_COMPONENT_SCANNER,
|
SizeResolution::Unknown { physical, .. } => {
|
||||||
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
|
self.heal_actions(oi, *physical, size_summary).await;
|
||||||
bucket = %self.bucket,
|
size_summary.actions_accounting_unknown(oi);
|
||||||
object = %oi.name,
|
continue;
|
||||||
state = "size_lookup_failed",
|
}
|
||||||
"Scanner lifecycle action used fallback size"
|
SizeResolution::Corrupt { .. } => {
|
||||||
);
|
size_summary.actions_accounting_unknown(oi);
|
||||||
continue;
|
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);
|
self.alert_excessive_versions(object_infos.len(), cumulative_size);
|
||||||
@@ -400,25 +672,108 @@ impl ScannerItem {
|
|||||||
let mut to_delete_objs: Vec<ObjectToDelete> = Vec::new();
|
let mut to_delete_objs: Vec<ObjectToDelete> = Vec::new();
|
||||||
let mut noncurrent_events: Vec<Event> = Vec::new();
|
let mut noncurrent_events: Vec<Event> = Vec::new();
|
||||||
let mut noncurrent_accounting: Vec<PendingScannerAccounting<'_>> = 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 cumulative_size = 0;
|
||||||
let mut remaining_versions = object_infos.len();
|
let mut remaining_versions = object_infos.len();
|
||||||
'eventLoop: {
|
'eventLoop: {
|
||||||
for (i, event) in events.iter().enumerate() {
|
for (i, event) in events.iter().enumerate() {
|
||||||
let oi = &object_infos[i];
|
let oi = &object_infos[i];
|
||||||
let actual_size = match oi.get_actual_size() {
|
let known_size = resolved_sizes[i].known_size();
|
||||||
Ok(size) => size,
|
if has_corrupt_size
|
||||||
Err(_) => {
|
&& matches!(
|
||||||
warn!(
|
event.action,
|
||||||
target: "rustfs::scanner::folder",
|
IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction
|
||||||
event = EVENT_SCANNER_LIFECYCLE_ACTION,
|
)
|
||||||
component = LOG_COMPONENT_SCANNER,
|
{
|
||||||
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
|
// An all-version delete would also remove a corrupt
|
||||||
bucket = %self.bucket,
|
// sibling that could not be reconciled safely.
|
||||||
object = %oi.name,
|
continue;
|
||||||
state = "size_lookup_failed",
|
}
|
||||||
"Scanner lifecycle action used fallback size"
|
if !lifecycle_event_allowed(&resolved_sizes[i], event, &lifecycle) {
|
||||||
);
|
// An unknown logical size must not make an otherwise
|
||||||
0
|
// 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 +801,24 @@ impl ScannerItem {
|
|||||||
done_ilm(1)();
|
done_ilm(1)();
|
||||||
remaining_versions = 0;
|
remaining_versions = 0;
|
||||||
} else {
|
} else {
|
||||||
PendingScannerAccounting {
|
if let Some(actual_size) = known_size {
|
||||||
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
|
|
||||||
}
|
|
||||||
};
|
|
||||||
PendingScannerAccounting {
|
PendingScannerAccounting {
|
||||||
object: retained,
|
object: oi,
|
||||||
retained_size,
|
retained_size: actual_size,
|
||||||
expired_size: 0,
|
expired_size: 0,
|
||||||
}
|
}
|
||||||
.apply(size_summary, &mut cumulative_size, false);
|
.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;
|
break 'eventLoop;
|
||||||
}
|
}
|
||||||
@@ -511,11 +854,13 @@ impl ScannerItem {
|
|||||||
version_id: opt.version_id,
|
version_id: opt.version_id,
|
||||||
..Default::default()
|
..Default::default()
|
||||||
});
|
});
|
||||||
noncurrent_accounting.push(PendingScannerAccounting {
|
if let Some(actual_size) = known_size {
|
||||||
object: oi,
|
noncurrent_accounting.push(PendingScannerAccounting {
|
||||||
retained_size: actual_size,
|
object: oi,
|
||||||
expired_size: 0,
|
retained_size: actual_size,
|
||||||
});
|
expired_size: 0,
|
||||||
|
});
|
||||||
|
}
|
||||||
account_now = false;
|
account_now = false;
|
||||||
}
|
}
|
||||||
noncurrent_events.push(event.clone());
|
noncurrent_events.push(event.clone());
|
||||||
@@ -548,7 +893,7 @@ impl ScannerItem {
|
|||||||
|
|
||||||
if account_now {
|
if account_now {
|
||||||
size_summary.actions_accounting(oi, size, actual_size);
|
size_summary.actions_accounting(oi, size, actual_size);
|
||||||
cumulative_size += size;
|
cumulative_size = cumulative_size.saturating_add(size);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -576,11 +921,20 @@ impl ScannerItem {
|
|||||||
}
|
}
|
||||||
if record_scanner_ilm_action_if_queued(global_metrics(), action, count, queued) {
|
if record_scanner_ilm_action_if_queued(global_metrics(), action, count, queued) {
|
||||||
done_ilm(count)();
|
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 {
|
for pending in noncurrent_accounting {
|
||||||
pending.apply(size_summary, &mut cumulative_size, queued);
|
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);
|
self.alert_excessive_versions(remaining_versions, cumulative_size);
|
||||||
}
|
}
|
||||||
@@ -929,4 +1283,394 @@ mod tests {
|
|||||||
assert_eq!(item.object_name, "object");
|
assert_eq!(item.object_name, "object");
|
||||||
assert_eq!(item.object_path(), "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::Corrupt {
|
||||||
|
physical: 12,
|
||||||
|
reason: SizeResolutionReason::InvalidDeclaredSize,
|
||||||
|
},
|
||||||
|
&Event {
|
||||||
|
action: IlmAction::DeleteAction,
|
||||||
|
rule_id: "size".to_string(),
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
&size_filtered
|
||||||
|
));
|
||||||
|
assert!(lifecycle_rule_has_size_filter(
|
||||||
|
&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: None,
|
||||||
|
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()
|
||||||
|
},
|
||||||
|
""
|
||||||
|
));
|
||||||
|
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);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -326,6 +326,9 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
|
|||||||
skip_heal: Arc::new(AtomicBool::new(false)),
|
skip_heal: Arc::new(AtomicBool::new(false)),
|
||||||
local_disk: disk,
|
local_disk: disk,
|
||||||
pending_heals_changed: false,
|
pending_heals_changed: false,
|
||||||
|
pending_size_reconciliation_keys: HashSet::new(),
|
||||||
|
pending_size_reconciliation_scopes: HashSet::new(),
|
||||||
|
pending_size_reconciliation_truncated: false,
|
||||||
list_path_raw_options_observer: None,
|
list_path_raw_options_observer: None,
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -388,6 +391,66 @@ async fn test_record_failed_ttl_zero_noop() {
|
|||||||
assert!(!scanner.should_skip_failed("path2"));
|
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]
|
#[test]
|
||||||
fn test_classify_get_size_failure_marks_metadata_heal_object_path() {
|
fn test_classify_get_size_failure_marks_metadata_heal_object_path() {
|
||||||
let temp_dir = std::env::temp_dir();
|
let temp_dir = std::env::temp_dir();
|
||||||
|
|||||||
Reference in New Issue
Block a user