mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-05 19:55:37 +00:00
Compare commits
20 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 437b494552 | |||
| 41a1d9d681 | |||
| 5d9d88e9e9 | |||
| 466741449d | |||
| 7983a26269 | |||
| 043542a49b | |||
| 37667edf48 | |||
| b9e7fcd755 | |||
| e41ff01c68 | |||
| e907ae4be4 | |||
| 21103b320f | |||
| b51f46cf73 | |||
| 18d8a5e08e | |||
| 5c30c6cfac | |||
| 141c18d4af | |||
| 807d2a61a6 | |||
| 0438926adc | |||
| 4878caa36f | |||
| 40ebb0bd54 | |||
| 90ca02b27d |
@@ -782,6 +782,29 @@ impl DataUsageCache {
|
||||
(visited == expected_entries).then_some(entry)
|
||||
}
|
||||
|
||||
pub(crate) fn has_complete_root_inventory(&self, bucket_keys: &HashSet<String>) -> bool {
|
||||
let Some(root) = self.find(DATA_USAGE_ROOT) else {
|
||||
return false;
|
||||
};
|
||||
// Set roots only connect bucket entries. Scalar data at the root, an
|
||||
// extra bucket, or an orphan must not disappear during bucket folding.
|
||||
root.children.len() == bucket_keys.len()
|
||||
&& bucket_keys.iter().all(|key| root.children.contains(key))
|
||||
&& root.size == 0
|
||||
&& root.objects == 0
|
||||
&& root.versions == 0
|
||||
&& root.delete_markers == 0
|
||||
&& root.failed_objects == 0
|
||||
&& !root.compacted
|
||||
&& root.obj_sizes.is_empty()
|
||||
&& root.obj_versions.is_empty()
|
||||
&& root.replication_stats.is_none()
|
||||
&& root.all_tier_stats.is_none()
|
||||
&& root.unknown_tier_stats.is_none()
|
||||
&& root.tier_accounting_proof.is_none()
|
||||
&& self.checked_flatten_complete(DATA_USAGE_ROOT).is_some()
|
||||
}
|
||||
|
||||
fn checked_flatten_inner(&self, path: &str) -> Option<(DataUsageEntry, usize)> {
|
||||
let root_key = hash_path(path).key();
|
||||
let (root_key, root) = self.cache.get_key_value(&root_key)?;
|
||||
|
||||
@@ -4869,6 +4869,28 @@ async fn usage_bootstrap_does_not_overwrite_concurrent_replacement() {
|
||||
#[serial]
|
||||
async fn scanner_usage_state_reset_publishes_fenced_bootstrap_marker() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
let quota_ledger_path = "config/quota-ledger/reserved-bucket.json";
|
||||
let quota_ledger = serde_json::to_vec(&serde_json::json!({
|
||||
"version": 1,
|
||||
"bucket_incarnation": "00000000-0000-0000-0000-000000000001",
|
||||
"quota_revision_unix_nanos": 1,
|
||||
"accounted_usage": 100,
|
||||
"reservations": {
|
||||
"00000000-0000-0000-0000-000000000002": {
|
||||
"object": "pending-object",
|
||||
"old_size": 0,
|
||||
"new_size": 64,
|
||||
"created_at": 1,
|
||||
"pool_index": 0,
|
||||
"set_index": 0,
|
||||
"commit_started": true
|
||||
}
|
||||
}
|
||||
}))
|
||||
.expect("quota ledger fixture should encode");
|
||||
save_config(store.clone(), quota_ledger_path, quota_ledger.clone())
|
||||
.await
|
||||
.expect("independent quota reservations should persist");
|
||||
let cycle = CurrentCycle {
|
||||
current: 41,
|
||||
next: 42,
|
||||
@@ -4927,6 +4949,14 @@ async fn scanner_usage_state_reset_publishes_fenced_bootstrap_marker() {
|
||||
assert!(!data_usage_info_has_persisted_baseline_identity(&usage));
|
||||
assert_eq!(usage.scanner_epoch, Some(9));
|
||||
|
||||
assert_eq!(
|
||||
read_config(store.clone(), quota_ledger_path)
|
||||
.await
|
||||
.expect("quota ledger must remain readable after scanner reset"),
|
||||
quota_ledger,
|
||||
"scanner reset must preserve incarnation and outstanding reserved bytes exactly"
|
||||
);
|
||||
|
||||
for path in [
|
||||
usage_backup_path.as_str(),
|
||||
LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(),
|
||||
|
||||
@@ -314,6 +314,30 @@ fn scanner_bucket_plan_digest(buckets: &[BucketInfo], activity_digest: [u8; 32])
|
||||
DataUsageScanPlanDigest(hasher.finalize().into())
|
||||
}
|
||||
|
||||
fn scanner_bucket_inventory_is_complete(
|
||||
all_buckets: &[BucketInfo],
|
||||
buckets_by_source: &HashMap<DataUsageCacheSource, Vec<BucketInfo>>,
|
||||
) -> bool {
|
||||
let inventory = all_buckets
|
||||
.iter()
|
||||
.map(|bucket| (bucket.name.as_str(), bucket.created))
|
||||
.collect::<HashMap<_, _>>();
|
||||
if inventory.len() != all_buckets.len() || inventory.keys().any(|name| name.is_empty() || *name == DATA_USAGE_ROOT) {
|
||||
return false;
|
||||
}
|
||||
let mut covered = HashSet::with_capacity(inventory.len());
|
||||
for buckets in buckets_by_source.values() {
|
||||
let mut set_names = HashSet::with_capacity(buckets.len());
|
||||
for bucket in buckets {
|
||||
if !set_names.insert(bucket.name.as_str()) || inventory.get(bucket.name.as_str()) != Some(&bucket.created) {
|
||||
return false;
|
||||
}
|
||||
covered.insert(bucket.name.as_str());
|
||||
}
|
||||
}
|
||||
covered.len() == inventory.len()
|
||||
}
|
||||
|
||||
fn scanner_bucket_cache_digest(
|
||||
scan_plan_digest: DataUsageScanPlanDigest,
|
||||
dirty_generation: Option<u64>,
|
||||
|
||||
@@ -213,10 +213,85 @@ pub(super) fn cache_snapshot_is_current(
|
||||
)
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub(super) struct ScannerSnapshotIdentity {
|
||||
pub(super) cycle: u64,
|
||||
pub(super) leader_epoch: u64,
|
||||
pub(super) plan_digest: DataUsageScanPlanDigest,
|
||||
pub(super) tier_registry_generation: Option<u64>,
|
||||
}
|
||||
|
||||
pub(super) struct ScannerSnapshotScope<'a> {
|
||||
pub(super) sources: &'a HashSet<DataUsageCacheSource>,
|
||||
pub(super) buckets: &'a [String],
|
||||
pub(super) identity: ScannerSnapshotIdentity,
|
||||
}
|
||||
|
||||
#[derive(Debug, PartialEq, Eq, thiserror::Error)]
|
||||
pub(super) enum ScannerSnapshotValidationError {
|
||||
#[error("scanner snapshot does not cover the expected complete sets")]
|
||||
IncompleteSets,
|
||||
#[error("scanner snapshot does not match the requested generation")]
|
||||
GenerationMismatch,
|
||||
#[error("scanner snapshot bucket inventory is invalid")]
|
||||
InvalidInventory,
|
||||
#[error("scanner snapshot root is incomplete or corrupt")]
|
||||
InvalidRoot,
|
||||
}
|
||||
|
||||
struct ValidatedScannerSnapshot<'a> {
|
||||
results: &'a [DataUsageCache],
|
||||
last_update: SystemTime,
|
||||
}
|
||||
|
||||
impl<'a> ValidatedScannerSnapshot<'a> {
|
||||
fn validate(
|
||||
results: &'a [DataUsageCache],
|
||||
scope: &ScannerSnapshotScope<'_>,
|
||||
) -> std::result::Result<Self, ScannerSnapshotValidationError> {
|
||||
if !scanner_results_form_complete_snapshot(results, scope.sources) {
|
||||
return Err(ScannerSnapshotValidationError::IncompleteSets);
|
||||
}
|
||||
let bucket_keys = scope
|
||||
.buckets
|
||||
.iter()
|
||||
.map(|bucket| crate::hash_path(bucket).key())
|
||||
.collect::<HashSet<_>>();
|
||||
if bucket_keys.len() != scope.buckets.len()
|
||||
|| scope
|
||||
.buckets
|
||||
.iter()
|
||||
.any(|bucket| bucket.is_empty() || bucket == DATA_USAGE_ROOT)
|
||||
{
|
||||
return Err(ScannerSnapshotValidationError::InvalidInventory);
|
||||
}
|
||||
for result in results {
|
||||
if result.info.next_cycle != scope.identity.cycle
|
||||
|| result.info.leader_epoch != scope.identity.leader_epoch
|
||||
|| result.info.scan_plan_digest != Some(scope.identity.plan_digest)
|
||||
|| result.info.tier_registry_generation != scope.identity.tier_registry_generation
|
||||
{
|
||||
return Err(ScannerSnapshotValidationError::GenerationMismatch);
|
||||
}
|
||||
if result.info.name != DATA_USAGE_ROOT
|
||||
|| result.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT
|
||||
|| !result.has_complete_root_inventory(&bucket_keys)
|
||||
{
|
||||
return Err(ScannerSnapshotValidationError::InvalidRoot);
|
||||
}
|
||||
}
|
||||
let last_update = results
|
||||
.iter()
|
||||
.filter_map(|result| result.info.last_update)
|
||||
.max()
|
||||
.ok_or(ScannerSnapshotValidationError::IncompleteSets)?;
|
||||
Ok(Self { results, last_update })
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn completed_data_usage_info(
|
||||
results: &[DataUsageCache],
|
||||
expected_sources: &HashSet<DataUsageCacheSource>,
|
||||
all_buckets: &[String],
|
||||
scope: &ScannerSnapshotScope<'_>,
|
||||
tier_registry_names: &[String],
|
||||
bucket_plan_complete: bool,
|
||||
budget_elapsed: bool,
|
||||
@@ -229,26 +304,10 @@ pub(super) fn completed_data_usage_info(
|
||||
if !should_publish_completed_snapshot(completed_set_count, results.len(), budget_elapsed, cancelled) {
|
||||
return None;
|
||||
}
|
||||
if !scanner_results_form_complete_snapshot(results, expected_sources) {
|
||||
return None;
|
||||
}
|
||||
|
||||
// A generation is comparable across nodes because it is derived from the
|
||||
// frozen registry names. Cycle and leader fencing remain separate cache
|
||||
// metadata. Legacy peers omit the generation; an all-legacy result remains
|
||||
// readable, but mixing legacy and new (or two new generations) would make
|
||||
// the per-tier accounting ambiguous.
|
||||
let registry_generation = results.first()?.info.tier_registry_generation;
|
||||
if results.iter().any(|result| match registry_generation {
|
||||
Some(generation) => result.info.tier_registry_generation != Some(generation),
|
||||
None => result.info.tier_registry_generation.is_some(),
|
||||
}) {
|
||||
return None;
|
||||
}
|
||||
|
||||
if results.iter().any(|result| result.root().is_none()) {
|
||||
return None;
|
||||
}
|
||||
let validated = ValidatedScannerSnapshot::validate(results, scope).ok()?;
|
||||
let results = validated.results;
|
||||
let all_buckets = scope.buckets;
|
||||
let registry_generation = scope.identity.tier_registry_generation;
|
||||
|
||||
let mut total = DataUsageEntry::default();
|
||||
let mut bucket_entries = HashMap::with_capacity(all_buckets.len());
|
||||
@@ -273,7 +332,7 @@ pub(super) fn completed_data_usage_info(
|
||||
return None;
|
||||
}
|
||||
|
||||
let merged_last_update = results.iter().filter_map(|result| result.info.last_update).max()?;
|
||||
let merged_last_update = validated.last_update;
|
||||
let buckets_usage = bucket_entries
|
||||
.iter()
|
||||
.map(|(bucket, entry)| Some((bucket.clone(), checked_bucket_usage_info(entry)?)))
|
||||
@@ -300,8 +359,8 @@ pub(super) fn completed_data_usage_info(
|
||||
usage_snapshot_set_states.sort_by_key(|state| (state.pool_index, state.set_index));
|
||||
let data_usage_info = DataUsageInfo {
|
||||
last_update: Some(merged_last_update),
|
||||
scanner_cycle: Some(results.first()?.info.next_cycle),
|
||||
scanner_epoch: Some(results.first()?.info.leader_epoch),
|
||||
scanner_cycle: Some(scope.identity.cycle),
|
||||
scanner_epoch: Some(scope.identity.leader_epoch),
|
||||
objects_total_count: u64::try_from(total.objects).ok()?,
|
||||
versions_total_count: u64::try_from(total.versions).ok()?,
|
||||
delete_markers_total_count: u64::try_from(total.delete_markers).ok()?,
|
||||
|
||||
@@ -39,6 +39,12 @@ pub(super) fn prepare_scoped_set_scan(
|
||||
else {
|
||||
return None;
|
||||
};
|
||||
// The existing cache does not bind each bucket to a durable incarnation.
|
||||
// Listing creation times can come from volume metadata, so even Some(time)
|
||||
// cannot prove that an unselected same-name bucket is the cached bucket.
|
||||
if all_buckets.iter().any(|bucket| !selected_buckets.contains(&bucket.name)) {
|
||||
return None;
|
||||
}
|
||||
if selected_buckets.is_empty()
|
||||
|| !old_cache.info.snapshot_complete
|
||||
|| old_cache.info.last_update.is_none()
|
||||
@@ -49,7 +55,7 @@ pub(super) fn prepare_scoped_set_scan(
|
||||
|| old_cache.info.source != Some(generation.source)
|
||||
|| old_cache.info.scan_plan_digest != Some(baseline_scan_plan_digest)
|
||||
|| old_cache.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT
|
||||
|| old_cache.checked_flatten_complete_scope(DATA_USAGE_ROOT).is_none()
|
||||
|| !old_cache.has_complete_root_inventory(&old_cache.find(DATA_USAGE_ROOT)?.children)
|
||||
{
|
||||
return None;
|
||||
}
|
||||
@@ -74,21 +80,12 @@ pub(super) fn prepare_scoped_set_scan(
|
||||
cache: HashMap::new(),
|
||||
};
|
||||
cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
|
||||
let root_hash = crate::hash_path(DATA_USAGE_ROOT);
|
||||
let mut current_bucket_names = HashSet::with_capacity(all_buckets.len());
|
||||
for bucket in all_buckets {
|
||||
if !current_bucket_names.insert(bucket.name.as_str()) {
|
||||
return None;
|
||||
}
|
||||
if selected_buckets.contains(&bucket.name) {
|
||||
cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
|
||||
continue;
|
||||
}
|
||||
|
||||
let bucket_hash = crate::hash_path(&bucket.name);
|
||||
old_cache.find(&bucket.name)?;
|
||||
cache.copy_with_children(old_cache, &bucket_hash, &Some(root_hash.clone()));
|
||||
cache.find(&bucket.name)?;
|
||||
cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
|
||||
}
|
||||
|
||||
Some(PreparedScopedSetScan {
|
||||
|
||||
@@ -260,6 +260,7 @@ where
|
||||
}
|
||||
}
|
||||
bucket_plan_complete &= buckets_by_source.keys().copied().collect::<HashSet<_>>() == *expected_sources;
|
||||
bucket_plan_complete &= scanner_bucket_inventory_is_complete(&all_buckets, &buckets_by_source);
|
||||
let activity_digest = crate::scanner::scanner_activity_snapshot_digest(&activity_before);
|
||||
let scan_plan_digest =
|
||||
scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_structural_digest(&activity_before));
|
||||
@@ -543,8 +544,16 @@ where
|
||||
let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::<Vec<_>>();
|
||||
let completed_usage = completed_data_usage_info(
|
||||
&results,
|
||||
&expected_sources,
|
||||
&all_bucket_names,
|
||||
&ScannerSnapshotScope {
|
||||
sources: &expected_sources,
|
||||
buckets: &all_bucket_names,
|
||||
identity: ScannerSnapshotIdentity {
|
||||
cycle: want_cycle,
|
||||
leader_epoch,
|
||||
plan_digest: scan_plan_digest,
|
||||
tier_registry_generation: Some(tier_registry_generation),
|
||||
},
|
||||
},
|
||||
&tier_registry.names,
|
||||
bucket_plan_complete,
|
||||
budget_elapsed,
|
||||
|
||||
@@ -18,6 +18,32 @@ use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage, TierAccount
|
||||
|
||||
const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]);
|
||||
|
||||
#[test]
|
||||
fn scanner_bucket_inventory_requires_exact_unique_set_union() {
|
||||
let first = BucketInfo {
|
||||
name: "first".to_string(),
|
||||
..Default::default()
|
||||
};
|
||||
let second = BucketInfo {
|
||||
name: "second".to_string(),
|
||||
..Default::default()
|
||||
};
|
||||
let source = DataUsageCacheSource::new(0, 0);
|
||||
let mut sets = HashMap::from([(source, vec![first.clone()])]);
|
||||
assert!(scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets));
|
||||
assert!(!scanner_bucket_inventory_is_complete(&[first.clone(), second.clone()], &sets));
|
||||
assert!(!scanner_bucket_inventory_is_complete(&[], &sets));
|
||||
assert!(!scanner_bucket_inventory_is_complete(&[first.clone(), first.clone()], &sets));
|
||||
sets.insert(source, vec![first.clone(), first.clone()]);
|
||||
assert!(!scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets));
|
||||
sets.insert(source, vec![second]);
|
||||
assert!(!scanner_bucket_inventory_is_complete(std::slice::from_ref(&first), &sets));
|
||||
let mut recreated = first.clone();
|
||||
recreated.created = Some(OffsetDateTime::UNIX_EPOCH);
|
||||
sets.insert(source, vec![recreated]);
|
||||
assert!(!scanner_bucket_inventory_is_complete(&[first], &sets));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn should_publish_completed_snapshot_requires_full_clean_cycle() {
|
||||
assert!(should_publish_completed_snapshot(3, 3, false, false));
|
||||
@@ -108,7 +134,133 @@ fn completed_data_usage_info_for_test(
|
||||
cancelled: bool,
|
||||
) -> Option<(DataUsageInfo, SystemTime)> {
|
||||
let expected_sources = results.iter().filter_map(|result| result.info.source).collect::<HashSet<_>>();
|
||||
completed_data_usage_info(results, &expected_sources, all_buckets, &[], true, budget_elapsed, cancelled)
|
||||
completed_usage_for_scope(results, &expected_sources, all_buckets, &[], true, budget_elapsed, cancelled)
|
||||
}
|
||||
|
||||
fn completed_usage_for_scope(
|
||||
results: &[DataUsageCache],
|
||||
expected_sources: &HashSet<DataUsageCacheSource>,
|
||||
all_buckets: &[String],
|
||||
tier_registry_names: &[String],
|
||||
bucket_plan_complete: bool,
|
||||
budget_elapsed: bool,
|
||||
cancelled: bool,
|
||||
) -> Option<(DataUsageInfo, SystemTime)> {
|
||||
let first = results.first()?;
|
||||
completed_data_usage_info(
|
||||
results,
|
||||
&ScannerSnapshotScope {
|
||||
sources: expected_sources,
|
||||
buckets: all_buckets,
|
||||
identity: ScannerSnapshotIdentity {
|
||||
cycle: first.info.next_cycle,
|
||||
leader_epoch: first.info.leader_epoch,
|
||||
plan_digest: TEST_PLAN_DIGEST,
|
||||
tier_registry_generation: first.info.tier_registry_generation,
|
||||
},
|
||||
},
|
||||
tier_registry_names,
|
||||
bucket_plan_complete,
|
||||
budget_elapsed,
|
||||
cancelled,
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completed_data_usage_info_rejects_duplicate_bucket_inventory() {
|
||||
let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
|
||||
let buckets = vec!["bucket".to_string(), "bucket".to_string()];
|
||||
assert!(completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completed_data_usage_info_rejects_extra_or_detached_bucket_data() {
|
||||
let buckets = vec!["bucket".to_string()];
|
||||
let mut set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
|
||||
set.replace(
|
||||
"unlisted",
|
||||
DATA_USAGE_ROOT,
|
||||
DataUsageEntry {
|
||||
objects: 1,
|
||||
..Default::default()
|
||||
},
|
||||
);
|
||||
assert!(completed_data_usage_info_for_test(&[set.clone()], &buckets, false, false).is_none());
|
||||
set.cache
|
||||
.get_mut(DATA_USAGE_ROOT)
|
||||
.expect("set root")
|
||||
.children
|
||||
.remove(&hash_path("unlisted").key());
|
||||
assert!(
|
||||
completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none(),
|
||||
"orphaned data must not disappear from authoritative accounting"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completed_data_usage_info_rejects_disconnected_expected_bucket() {
|
||||
let buckets = vec!["bucket".to_string()];
|
||||
let mut set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
|
||||
set.cache.get_mut(DATA_USAGE_ROOT).expect("set root").children.clear();
|
||||
assert!(completed_data_usage_info_for_test(&[set], &buckets, false, false).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completed_data_usage_info_rejects_root_scalar_data_and_unknown_key_format() {
|
||||
let buckets = vec!["bucket".to_string()];
|
||||
let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
|
||||
let mut scalar_root = set.clone();
|
||||
scalar_root.cache.get_mut(DATA_USAGE_ROOT).expect("set root").size = 10;
|
||||
assert!(completed_data_usage_info_for_test(&[scalar_root], &buckets, false, false).is_none());
|
||||
let mut future_format = set;
|
||||
future_format.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT + 1;
|
||||
assert!(completed_data_usage_info_for_test(&[future_format], &buckets, false, false).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completed_data_usage_info_binds_all_results_to_requested_identity() {
|
||||
let buckets = vec!["bucket".to_string()];
|
||||
let source = DataUsageCacheSource::new(0, 0);
|
||||
let sources = HashSet::from([source]);
|
||||
let set = completed_root_cache("bucket", 2, 10, source);
|
||||
let identity = ScannerSnapshotIdentity {
|
||||
cycle: 0,
|
||||
leader_epoch: 0,
|
||||
plan_digest: TEST_PLAN_DIGEST,
|
||||
tier_registry_generation: None,
|
||||
};
|
||||
let results = [set];
|
||||
for expected in [
|
||||
ScannerSnapshotIdentity { cycle: 1, ..identity },
|
||||
ScannerSnapshotIdentity {
|
||||
leader_epoch: 1,
|
||||
..identity
|
||||
},
|
||||
ScannerSnapshotIdentity {
|
||||
plan_digest: DataUsageScanPlanDigest([9; 32]),
|
||||
..identity
|
||||
},
|
||||
ScannerSnapshotIdentity {
|
||||
tier_registry_generation: Some(1),
|
||||
..identity
|
||||
},
|
||||
] {
|
||||
let scope = ScannerSnapshotScope {
|
||||
sources: &sources,
|
||||
buckets: &buckets,
|
||||
identity: expected,
|
||||
};
|
||||
assert!(completed_data_usage_info(&results, &scope, &[], true, false, false).is_none());
|
||||
}
|
||||
let scope = ScannerSnapshotScope {
|
||||
sources: &sources,
|
||||
buckets: &buckets,
|
||||
identity,
|
||||
};
|
||||
let (usage, _) = completed_data_usage_info(&results, &scope, &[], true, false, false)
|
||||
.expect("the requested complete scope remains publishable");
|
||||
assert_eq!(usage.objects_total_count, 2);
|
||||
assert!(usage.is_complete_bucket_usage_snapshot());
|
||||
}
|
||||
|
||||
fn lkg_root_cache(bucket: &str, objects: usize, source: DataUsageCacheSource) -> DataUsageCache {
|
||||
@@ -136,7 +288,7 @@ fn partial_usage_is_observational_not_authoritative_for_quota() {
|
||||
let expected = HashSet::from([current_source, stalled_source]);
|
||||
|
||||
assert!(
|
||||
completed_data_usage_info(&[current.clone(), stalled.clone()], &expected, &all_buckets, &[], true, false, false)
|
||||
completed_usage_for_scope(&[current.clone(), stalled.clone()], &expected, &all_buckets, &[], true, false, false)
|
||||
.is_none()
|
||||
);
|
||||
let (observed, _) = observational_data_usage_info(&[current, stalled], &expected, &all_buckets, &[], TEST_PLAN_DIGEST, 8, 3)
|
||||
@@ -509,7 +661,7 @@ fn completed_data_usage_info_accepts_unknown_only_with_current_registry_generati
|
||||
|
||||
let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);
|
||||
assert!(
|
||||
completed_data_usage_info(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_some()
|
||||
completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_some()
|
||||
);
|
||||
}
|
||||
|
||||
@@ -547,7 +699,7 @@ fn completed_data_usage_info_rejects_non_registry_tier_in_current_generation() {
|
||||
|
||||
let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);
|
||||
assert!(
|
||||
completed_data_usage_info(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_none()
|
||||
completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_none()
|
||||
);
|
||||
}
|
||||
|
||||
@@ -698,6 +850,7 @@ fn completed_data_usage_info_publishes_confirmed_empty_namespace() {
|
||||
source: Some(DataUsageCacheSource::new(0, 0)),
|
||||
snapshot_complete: true,
|
||||
scan_plan_digest: Some(TEST_PLAN_DIGEST),
|
||||
cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
|
||||
..Default::default()
|
||||
},
|
||||
..Default::default()
|
||||
@@ -855,7 +1008,7 @@ fn completed_data_usage_info_requires_exact_topology_sources() {
|
||||
let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0), DataUsageCacheSource::new(1, 0)]);
|
||||
|
||||
assert!(
|
||||
completed_data_usage_info(&[first_set, unexpected_set], &expected_sources, &all_buckets, &[], true, false, false)
|
||||
completed_usage_for_scope(&[first_set, unexpected_set], &expected_sources, &all_buckets, &[], true, false, false)
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
@@ -866,7 +1019,7 @@ fn completed_data_usage_info_rejects_incomplete_bucket_plan() {
|
||||
let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
|
||||
let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);
|
||||
|
||||
assert!(completed_data_usage_info(&[set], &expected_sources, &all_buckets, &[], false, false, false).is_none());
|
||||
assert!(completed_usage_for_scope(&[set], &expected_sources, &all_buckets, &[], false, false, false).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -914,6 +914,13 @@ fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsa
|
||||
cache
|
||||
}
|
||||
|
||||
fn bucket_info_with_created_time(name: &str) -> BucketInfo {
|
||||
BucketInfo {
|
||||
created: Some(time::OffsetDateTime::UNIX_EPOCH),
|
||||
..bucket_info(name)
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers() {
|
||||
@@ -1222,7 +1229,7 @@ fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
|
||||
fn scoped_set_scan_rebuilds_selected_buckets_and_drops_deleted_buckets() {
|
||||
let baseline_digest = DataUsageScanPlanDigest([1; 32]);
|
||||
let current_digest = DataUsageScanPlanDigest([2; 32]);
|
||||
let mut old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20), ("deleted", 30)], baseline_digest);
|
||||
@@ -1235,8 +1242,11 @@ fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
|
||||
..Default::default()
|
||||
},
|
||||
);
|
||||
let all_buckets = vec![bucket_info("stable"), bucket_info("dirty")];
|
||||
let selected_buckets = Arc::new(HashSet::from(["dirty".to_string(), "deleted".to_string()]));
|
||||
let all_buckets = vec![
|
||||
bucket_info_with_created_time("stable"),
|
||||
bucket_info_with_created_time("dirty"),
|
||||
];
|
||||
let selected_buckets = Arc::new(HashSet::from(["stable".to_string(), "dirty".to_string(), "deleted".to_string()]));
|
||||
|
||||
let prepared = prepare_scoped_set_scan(
|
||||
&old_cache,
|
||||
@@ -1256,12 +1266,16 @@ fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
|
||||
)
|
||||
.expect("complete matching set cache should support a scoped scan");
|
||||
|
||||
assert_eq!(prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::<Vec<_>>(), ["dirty"]);
|
||||
assert_eq!(
|
||||
prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::<Vec<_>>(),
|
||||
["stable", "dirty"]
|
||||
);
|
||||
let stable = prepared
|
||||
.cache
|
||||
.checked_flatten("stable")
|
||||
.expect("unselected bucket subtree should be retained");
|
||||
assert_eq!((stable.size, stable.objects), (15, 2));
|
||||
.expect("selected bucket placeholder should exist");
|
||||
assert_eq!((stable.size, stable.objects), (0, 0));
|
||||
assert!(prepared.cache.find("stable/prefix").is_none());
|
||||
assert_eq!(prepared.cache.find("dirty").map(|entry| (entry.size, entry.objects)), Some((0, 0)));
|
||||
assert!(prepared.cache.find("deleted").is_none());
|
||||
assert_eq!(prepared.cache.info.scan_plan_digest, Some(current_digest));
|
||||
@@ -1272,11 +1286,41 @@ fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
|
||||
assert_eq!(prepared.cache.info.lkg_scan_plan_digest, Some(baseline_digest));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scoped_set_scan_rejects_unbound_bucket_incarnations() {
|
||||
let baseline_digest = DataUsageScanPlanDigest([1; 32]);
|
||||
let old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20)], baseline_digest);
|
||||
let scope = ScannerBucketScanScope {
|
||||
selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
|
||||
baseline_scan_plan_digest: Some(baseline_digest),
|
||||
};
|
||||
let generation = ScannerSetCacheGeneration {
|
||||
want_cycle: 8,
|
||||
leader_epoch: 11,
|
||||
tier_registry_generation: 13,
|
||||
source: DataUsageCacheSource::new(1, 2),
|
||||
scan_plan_digest: DataUsageScanPlanDigest([2; 32]),
|
||||
};
|
||||
for created in [
|
||||
None,
|
||||
Some(OffsetDateTime::UNIX_EPOCH),
|
||||
Some(OffsetDateTime::UNIX_EPOCH + time::Duration::days(1)),
|
||||
] {
|
||||
let mut stable = bucket_info("stable");
|
||||
stable.created = created;
|
||||
let buckets = vec![stable, bucket_info_with_created_time("dirty")];
|
||||
assert!(
|
||||
prepare_scoped_set_scan(&old_cache, &buckets, &buckets, &scope, generation).is_none(),
|
||||
"missing identity, volume timestamps and same-name recreation must all rebuild"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() {
|
||||
let baseline_digest = DataUsageScanPlanDigest([3; 32]);
|
||||
let old_cache = complete_set_usage_cache(&[("stable", 10)], baseline_digest);
|
||||
let all_buckets = vec![bucket_info("stable"), bucket_info("new")];
|
||||
let all_buckets = vec![bucket_info_with_created_time("stable"), bucket_info_with_created_time("new")];
|
||||
|
||||
assert!(
|
||||
prepare_scoped_set_scan(
|
||||
@@ -1302,7 +1346,7 @@ fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() {
|
||||
#[test]
|
||||
fn scoped_set_scan_requires_an_exact_complete_baseline() {
|
||||
let baseline_digest = DataUsageScanPlanDigest([5; 32]);
|
||||
let all_buckets = vec![bucket_info("dirty")];
|
||||
let all_buckets = vec![bucket_info_with_created_time("dirty")];
|
||||
let scope = ScannerBucketScanScope {
|
||||
selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
|
||||
baseline_scan_plan_digest: Some(baseline_digest),
|
||||
@@ -1323,6 +1367,10 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() {
|
||||
not_durable.info.last_update = None;
|
||||
assert!(prepare_scoped_set_scan(¬_durable, &all_buckets, &all_buckets, &scope, generation).is_none());
|
||||
|
||||
let mut unscoped_usage = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
|
||||
unscoped_usage.cache.get_mut(DATA_USAGE_ROOT).expect("set root").objects = 1;
|
||||
assert!(prepare_scoped_set_scan(&unscoped_usage, &all_buckets, &all_buckets, &scope, generation).is_none());
|
||||
|
||||
let mut wrong_digest = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
|
||||
wrong_digest.info.scan_plan_digest = Some(DataUsageScanPlanDigest([7; 32]));
|
||||
assert!(prepare_scoped_set_scan(&wrong_digest, &all_buckets, &all_buckets, &scope, generation).is_none());
|
||||
@@ -1333,6 +1381,13 @@ fn scoped_set_scan_requires_an_exact_complete_baseline() {
|
||||
};
|
||||
let complete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
|
||||
assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &empty_scope, generation).is_none());
|
||||
assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &scope, generation).is_some());
|
||||
|
||||
let unidentified_buckets = vec![bucket_info("dirty")];
|
||||
assert!(
|
||||
prepare_scoped_set_scan(&complete, &unidentified_buckets, &unidentified_buckets, &scope, generation).is_some(),
|
||||
"fully selected buckets are rebuilt without reusing an unproven incarnation"
|
||||
);
|
||||
|
||||
let mut future_cache = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
|
||||
future_cache.info.next_cycle = generation.want_cycle.saturating_add(1);
|
||||
|
||||
@@ -50,8 +50,10 @@ cd "$(dirname "$0")/.."
|
||||
# s3_error! stays flat at 1616.
|
||||
# 1616 → 1613 on 2026-09-02: dependency refresh verified the current tree has
|
||||
# already shed three s3_error! invocation lines; retighten the line counter.
|
||||
# 1613 → 1600 on 2026-09-06: scanner publication coverage follow-up inherits
|
||||
# current s3_error! shrinkage; retighten the line counter.
|
||||
S3S_IMPORT_FILES_BASELINE=213
|
||||
S3_ERROR_LINES_BASELINE=1613
|
||||
S3_ERROR_LINES_BASELINE=1600
|
||||
# ecstore-scoped ratchet (rustfs/backlog#1842): the storage engine must not
|
||||
# know S3 wire/DTO types (ARCHITECTURE.md invariant 4). The S3-*consuming*
|
||||
# client was extracted to crates/s3-client, where s3s usage is legitimate;
|
||||
|
||||
Reference in New Issue
Block a user