diff --git a/.config/nextest.toml b/.config/nextest.toml index 5e50b639a..544783bc9 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -29,6 +29,7 @@ [test-groups] ecstore-serial-flaky = { max-threads = 1 } +embedded-test-ports = { max-threads = 1 } # Reliability / fault-injection e2e tests each spawn a single-node 4-disk RustFS # server and manipulate its disk directories at runtime (crates/e2e_test: @@ -54,6 +55,13 @@ test-group = 'ecstore-serial-flaky' filter = 'package(rustfs-ecstore) & test(/^set_disk::ops::multipart::tests::crash_consistency::/)' test-group = 'ecstore-serial-flaky' +# Embedded integration-test binaries discover an ephemeral port and release +# the probe listener before RustFS binds it. Serialize that cross-process +# TOCTOU window; retries would only hide real startup failures. +[[profile.default.overrides]] +filter = 'package(rustfs) & (binary(/^embedded.*_test$/) | binary(admin_diagnostic_capability_e2e))' +test-group = 'embedded-test-ports' + # Serialize the durable manual-transition checkpoint test across nextest's # process boundary; it mutates bucket lifecycle metadata and is not quarantined. [[profile.default.overrides]] @@ -143,6 +151,12 @@ test-group = 'e2e-reliability' filter = 'package(rustfs-ecstore) & test(/^set_disk::ops::multipart::tests::crash_consistency::/)' test-group = 'ecstore-serial-flaky' +# Match the default-profile embedded test isolation without quarantining or +# retrying failures in CI. +[[profile.ci.overrides]] +filter = 'package(rustfs) & (binary(/^embedded.*_test$/) | binary(admin_diagnostic_capability_e2e))' +test-group = 'embedded-test-ports' + # Serialize the durable manual-transition checkpoint test under the ci profile # too. No retries: failures stay visible. [[profile.ci.overrides]] diff --git a/crates/data-usage/src/data_usage.rs b/crates/data-usage/src/data_usage.rs index a0a368c91..d450adede 100644 --- a/crates/data-usage/src/data_usage.rs +++ b/crates/data-usage/src/data_usage.rs @@ -37,6 +37,10 @@ pub const USAGE_LAST_UPDATE_FUTURE_TOLERANCE: Duration = Duration::from_secs(5 * /// Keeping the existing object name preserves rolling-upgrade and rollback /// compatibility without allowing an ambiguous snapshot to become authoritative. pub const DATA_USAGE_OBJECT_NAME: &str = ".usage.v2.json"; +/// Latest structurally complete scanner observation. Unlike +/// [`DATA_USAGE_OBJECT_NAME`], this object is never authoritative for quota +/// admission because namespace activity may have raced the scan. +pub const DATA_USAGE_OBSERVED_OBJECT_NAME: &str = ".usage.observed.json"; /// Usage snapshot written by scanner implementations predating distributed /// leadership fencing. It is read only when neither authoritative snapshot @@ -218,6 +222,20 @@ pub struct DataUsageInfo { /// explicit entry for every bucket, including confirmed-empty buckets. #[serde(default)] pub usage_snapshot_complete: bool, + /// Whether no namespace activity or dirty-usage generation changed while + /// the coordinated snapshot was being produced. + /// + /// `false` still describes a structurally complete, useful point-in-time + /// usage view, but follow-up scanner work remains pending. `None` is kept + /// for snapshots written before this status became observable. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub usage_snapshot_converged: Option, + /// Identity of the authoritative snapshot from which a nonconverged + /// observation started. Admin readers require an exact match before using + /// the observation, so bucket namespace mutations fence old observations + /// without relying on synchronized clocks. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub usage_snapshot_authoritative_baseline: Option, /// Deprecated kept here for backward compatibility reasons pub bucket_sizes: HashMap, /// Per-disk snapshot information when available @@ -225,6 +243,59 @@ pub struct DataUsageInfo { pub disk_usage_status: Vec, } +/// Stable identity fields changed by both coordinated scanner publication and +/// backward-compatible bucket namespace cleanup. +#[derive(Debug, Default, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +pub struct DataUsageSnapshotIdentity { + pub last_update: Option, + pub scanner_cycle: Option, + pub scanner_epoch: Option, +} + +impl DataUsageInfo { + pub fn snapshot_identity(&self) -> DataUsageSnapshotIdentity { + DataUsageSnapshotIdentity { + last_update: self.last_update, + scanner_cycle: self.scanner_cycle, + scanner_epoch: self.scanner_epoch, + } + } +} + +/// Return whether `candidate` was produced after `baseline`. +/// +/// New coordinated snapshots are ordered by leadership epoch and scanner +/// cycle. The timestamp fallback preserves ordering for legacy snapshots that +/// predate those fields. +pub fn data_usage_snapshot_is_newer(candidate: &DataUsageInfo, baseline: &DataUsageInfo) -> bool { + match ( + candidate.scanner_epoch.zip(candidate.scanner_cycle), + baseline.scanner_epoch.zip(baseline.scanner_cycle), + ) { + (Some(candidate), Some(baseline)) => candidate > baseline, + (Some(_), None) => true, + (None, Some(_)) => false, + (None, None) => match (candidate.last_update, baseline.last_update) { + (Some(candidate), Some(baseline)) => candidate > baseline, + (Some(_), None) => true, + (None, Some(_) | None) => false, + }, + } +} + +/// Return whether a nonconverged observation may safely supersede the admin +/// view of `authoritative`. +/// +/// The exact baseline identity is independent of clock ordering. Older binaries +/// already advance the authoritative timestamp when deleting a bucket, so a +/// rollback delete/recreate fences the previous bucket incarnation too. +pub fn observed_data_usage_is_newer(observed: &DataUsageInfo, authoritative: &DataUsageInfo) -> bool { + observed.usage_snapshot_converged == Some(false) + && observed.is_complete_bucket_usage_snapshot() + && observed.usage_snapshot_authoritative_baseline.as_ref() == Some(&authoritative.snapshot_identity()) + && data_usage_snapshot_is_newer(observed, authoritative) +} + /// Metadata describing the status of a disk-level data usage snapshot. #[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct DiskUsageStatus { @@ -1783,6 +1854,8 @@ mod tests { let current = DataUsageInfo { last_update: Some(SystemTime::UNIX_EPOCH), usage_snapshot_complete: true, + usage_snapshot_converged: Some(false), + usage_snapshot_authoritative_baseline: Some(DataUsageSnapshotIdentity::default()), ..Default::default() }; let encoded = rmp_serde::to_vec_named(¤t).expect("encode current data usage snapshot"); @@ -1790,6 +1863,76 @@ mod tests { assert_eq!(legacy.buckets_count, 0); assert!(current.is_complete_bucket_usage_snapshot()); + assert_eq!(current.usage_snapshot_converged, Some(false)); + } + + #[test] + fn convergence_marker_defaults_to_unknown_for_older_snapshots() { + let encoded = rmp_serde::to_vec_named(&DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH), + usage_snapshot_complete: true, + ..Default::default() + }) + .expect("encode pre-convergence data usage snapshot"); + let decoded: DataUsageInfo = rmp_serde::from_slice(&encoded).expect("decode older data usage snapshot"); + + assert!(decoded.is_complete_bucket_usage_snapshot()); + assert_eq!(decoded.usage_snapshot_converged, None); + } + + #[test] + fn observation_selection_is_clock_independent_and_baseline_fenced() { + let mut authoritative = DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(600)), + scanner_epoch: Some(7), + scanner_cycle: Some(10), + usage_snapshot_complete: true, + ..Default::default() + }; + let observed = DataUsageInfo { + // A newer leader may have a slower wall clock. + last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(300)), + scanner_epoch: Some(8), + scanner_cycle: Some(1), + usage_snapshot_complete: true, + usage_snapshot_converged: Some(false), + usage_snapshot_authoritative_baseline: Some(authoritative.snapshot_identity()), + ..Default::default() + }; + + assert!(observed_data_usage_is_newer(&observed, &authoritative)); + + authoritative.last_update = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(601)); + assert!( + !observed_data_usage_is_newer(&observed, &authoritative), + "an old-binary namespace mutation must fence the prior bucket incarnation regardless of clock skew" + ); + } + + #[test] + fn observation_selection_requires_nonconverged_complete_newer_data() { + let authoritative = DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH), + scanner_epoch: Some(2), + scanner_cycle: Some(10), + usage_snapshot_complete: true, + ..Default::default() + }; + let baseline = Some(authoritative.snapshot_identity()); + let candidate = |epoch, cycle, converged, complete| DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1)), + scanner_epoch: Some(epoch), + scanner_cycle: Some(cycle), + usage_snapshot_complete: complete, + usage_snapshot_converged: converged, + usage_snapshot_authoritative_baseline: baseline, + ..Default::default() + }; + + assert!(observed_data_usage_is_newer(&candidate(2, 11, Some(false), true), &authoritative)); + assert!(!observed_data_usage_is_newer(&candidate(2, 9, Some(false), true), &authoritative)); + assert!(!observed_data_usage_is_newer(&candidate(2, 11, Some(true), true), &authoritative)); + assert!(!observed_data_usage_is_newer(&candidate(2, 11, Some(false), false), &authoritative)); } #[test] diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 59bf363f9..84caf7f26 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -310,7 +310,8 @@ pub mod config { pub mod data_usage { pub use crate::data_usage::{ DATA_USAGE_CACHE_NAME, apply_bucket_usage_memory_overlay, compute_bucket_usage, - init_compression_total_memory_from_backend, invalidate_data_usage_snapshot_cache, live_bucket_usage_computations, + init_compression_total_memory_from_backend, invalidate_admin_data_usage_snapshot_cache, + invalidate_data_usage_snapshot_cache, live_bucket_usage_computations, load_admin_data_usage_from_backend_cached, load_compression_total_from_memory, load_data_usage_from_backend, load_data_usage_from_backend_cached, record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory, record_bucket_object_write_memory, record_bucket_object_write_unknown_previous_memory, record_compression_total_memory, diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index d4f7364e3..6bb0811a4 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -20,7 +20,7 @@ pub mod local_snapshot; use crate::storage_api_contracts::{ bucket::{BucketOperations as _, BucketOptions}, list::{ListOperations as _, StorageListObjectVersionsInfo}, - object::{EcstoreObjectIO, HTTPPreconditions, ObjectIO as _}, + object::{EcstoreObjectIO, HTTPPreconditions, ObjectIO as _, ObjectOperations as _}, }; use crate::{ bucket::{metadata_sys::get_replication_config, versioning::VersioningApi as _, versioning_sys::BucketVersioningSys}, @@ -33,8 +33,9 @@ use crate::{ }; pub use local_snapshot::{LocalUsageSnapshot, read_snapshot as read_local_snapshot, snapshot_path}; use rustfs_data_usage::{ - BucketTargetUsageInfo, BucketUsageInfo, CompressionTotalInfo, DATA_USAGE_OBJECT_NAME, DataUsageCache, DataUsageEntry, - DataUsageInfo, DiskUsageStatus, LEGACY_DATA_USAGE_OBJECT_NAME, SizeHistogram, SizeSummary, VersionsHistogram, + BucketTargetUsageInfo, BucketUsageInfo, CompressionTotalInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME, + DataUsageCache, DataUsageEntry, DataUsageInfo, DiskUsageStatus, LEGACY_DATA_USAGE_OBJECT_NAME, SizeHistogram, SizeSummary, + VersionsHistogram, observed_data_usage_is_newer, }; use rustfs_io_metrics::record_system_path_failure; use rustfs_utils::path::SLASH_SEPARATOR; @@ -162,8 +163,9 @@ fn cache_data_usage_snapshot_result( result: Result<(DataUsageInfo, HashMap), Error>, loaded_at: tokio::time::Instant, refresh_generation: u64, + current_generation: u64, ) -> Option> { - if data_usage_snapshot_generation() != refresh_generation { + if current_generation != refresh_generation { return None; } @@ -196,6 +198,9 @@ type DataUsageSnapshotCache = Arc>>; static DATA_USAGE_SNAPSHOT_CACHE: OnceLock = OnceLock::new(); static DATA_USAGE_SNAPSHOT_REFRESH: OnceLock>> = OnceLock::new(); static DATA_USAGE_SNAPSHOT_GENERATION: AtomicU64 = AtomicU64::new(0); +static ADMIN_DATA_USAGE_SNAPSHOT_CACHE: OnceLock = OnceLock::new(); +static ADMIN_DATA_USAGE_SNAPSHOT_REFRESH: OnceLock>> = OnceLock::new(); +static ADMIN_DATA_USAGE_SNAPSHOT_GENERATION: AtomicU64 = AtomicU64::new(0); // Always-on revert detector for rustfs/backlog#1306: one relaxed increment per // full-bucket version listing is negligible and lets tests prove that admin @@ -254,11 +259,24 @@ fn data_usage_snapshot_generation() -> u64 { DATA_USAGE_SNAPSHOT_GENERATION.load(Ordering::Acquire) } +fn admin_data_usage_snapshot_cache() -> &'static DataUsageSnapshotCache { + ADMIN_DATA_USAGE_SNAPSHOT_CACHE.get_or_init(|| Arc::new(RwLock::new(None))) +} + +fn admin_data_usage_snapshot_generation() -> u64 { + ADMIN_DATA_USAGE_SNAPSHOT_GENERATION.load(Ordering::Acquire) +} + fn clear_data_usage_snapshot_cache(cache: &mut Option) { DATA_USAGE_SNAPSHOT_GENERATION.fetch_add(1, Ordering::AcqRel); *cache = None; } +fn clear_admin_data_usage_snapshot_cache(cache: &mut Option) { + ADMIN_DATA_USAGE_SNAPSHOT_GENERATION.fetch_add(1, Ordering::AcqRel); + *cache = None; +} + fn live_bucket_usage_cache() -> &'static LiveBucketUsageCache { LIVE_BUCKET_USAGE_CACHE.get_or_init(|| { moka::future::Cache::builder() @@ -283,6 +301,11 @@ lazy_static::lazy_static! { SLASH_SEPARATOR, DATA_USAGE_OBJECT_NAME ); + pub static ref DATA_USAGE_OBSERVED_OBJ_NAME_PATH: String = format!("{}{}{}", + crate::disk::BUCKET_META_PREFIX, + SLASH_SEPARATOR, + DATA_USAGE_OBSERVED_OBJECT_NAME + ); static ref DATA_USAGE_OBJ_BACKUP_PATH: String = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()); static ref LEGACY_DATA_USAGE_OBJ_NAME_PATH: String = format!("{}{}{}", crate::disk::BUCKET_META_PREFIX, @@ -357,6 +380,11 @@ fn stale_data_usage_persist_reason_for_source( /// Store data usage info to backend storage #[instrument(skip(store))] pub async fn store_data_usage_in_backend(data_usage_info: DataUsageInfo, store: Arc) -> Result<(), Error> { + if data_usage_info.usage_snapshot_converged == Some(false) { + return Err(Error::other( + "nonconverged data usage observations cannot replace the quota-authoritative snapshot", + )); + } // Prevent older data from overwriting newer persisted stats if let Ok((existing, source)) = load_data_usage_snapshot(store.clone()).await && source.is_authoritative() @@ -377,10 +405,12 @@ async fn save_data_usage_in_backend(data_usage_info: DataUsageInfo, store: Arc Result<(), Error>; +} + +#[async_trait::async_trait] +impl ObservedDataUsageSnapshotCleanup for ECStore { + async fn delete_observed_data_usage_snapshot(&self, revision: &str) -> Result<(), Error> { + self.delete_object( + RUSTFS_META_BUCKET, + DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + ObjectOptions { + delete_prefix: true, + delete_prefix_object: true, + http_preconditions: Some(HTTPPreconditions { + if_match: Some(revision.to_string()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + .map(|_| ()) + } +} + +async fn cleanup_observed_data_usage_after_authoritative_save(store: &S, authoritative: &DataUsageInfo) +where + S: EcstoreObjectIO + ObservedDataUsageSnapshotCleanup + ?Sized, +{ + let (observed, revision) = match load_data_usage_for_bucket_removal(store, DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()).await { + Ok(Some(snapshot)) => snapshot, + Ok(None) => return, + Err(err) => { + record_usage_snapshot_failure( + "read_observed_before_authoritative_cleanup", + DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + &err, + ); + return; + } + }; + if observed_data_usage_is_newer(&observed, authoritative) { + return; + } + + match store.delete_observed_data_usage_snapshot(&revision).await { + Ok(()) | Err(Error::ConfigNotFound | Error::FileNotFound | Error::ObjectNotFound(_, _) | Error::PreconditionFailed) => {} + Err(err) => { + record_usage_snapshot_failure( + "delete_observed_after_authoritative_save", + DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + &err, + ); + } + } +} + fn set_buckets_count_from_usage(data_usage_info: &mut DataUsageInfo) { data_usage_info.buckets_count = u64::try_from(data_usage_info.buckets_usage.len()).unwrap_or(u64::MAX); } @@ -443,6 +531,11 @@ pub(crate) async fn prepare_bucket_usage_for_namespace_change( let mut snapshot_cache = data_usage_snapshot_cache().write().await; ensure_bucket_namespace_guard(guard, bucket, "data usage snapshot cache cleanup")?; clear_data_usage_snapshot_cache(&mut snapshot_cache); + drop(snapshot_cache); + + let mut admin_snapshot_cache = admin_data_usage_snapshot_cache().write().await; + ensure_bucket_namespace_guard(guard, bucket, "admin data usage snapshot cache cleanup")?; + clear_admin_data_usage_snapshot_cache(&mut admin_snapshot_cache); Ok(()) } @@ -458,6 +551,10 @@ where let mut snapshot_cache = data_usage_snapshot_cache().write().await; ensure_bucket_namespace_guard(guard, bucket, "data usage snapshot cache invalidation")?; clear_data_usage_snapshot_cache(&mut snapshot_cache); + drop(snapshot_cache); + let mut admin_snapshot_cache = admin_data_usage_snapshot_cache().write().await; + ensure_bucket_namespace_guard(guard, bucket, "admin data usage snapshot cache invalidation")?; + clear_admin_data_usage_snapshot_cache(&mut admin_snapshot_cache); result } @@ -551,6 +648,23 @@ where ) .await?; + ensure_bucket_namespace_guard(guard, bucket, "observed data usage cleanup")?; + if let Err(err) = remove_bucket_usage_from_object_with_retries( + store, + DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + bucket, + DATA_USAGE_REMOVE_CAS_RETRIES, + None, + guard, + ) + .await + { + // The authoritative timestamp was already advanced above, so admin + // selection rejects this observation even if optional cleanup fails. + // Never make an admin-only freshness artifact block DeleteBucket. + record_usage_snapshot_failure("remove_bucket_from_observed", DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), &err); + } + for object in [ LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str(), LEGACY_DATA_USAGE_OBJ_BACKUP_PATH.as_str(), @@ -826,6 +940,61 @@ async fn load_data_usage_from_backend_with_baseline(store: Arc) -> Resu Ok(normalize_loaded_data_usage(data_usage_info, source.is_authoritative()).await) } +async fn load_observed_data_usage_snapshot(store: Arc) -> Option { + let data = match read_config_preserve_empty(store, &DATA_USAGE_OBSERVED_OBJ_NAME_PATH).await { + Ok(data) => data, + Err(Error::ConfigNotFound) => return None, + Err(err) => { + record_usage_snapshot_failure("read_observed", DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), &err); + return None; + } + }; + + match parse_usage_snapshot(&data) { + Ok(info) if info.usage_snapshot_converged == Some(false) && info.is_complete_bucket_usage_snapshot() => Some(info), + Ok(_) => { + error!( + event = "data_usage_snapshot_load_failed", + component = "ecstore", + subsystem = "data_usage", + state = "invalid_observed_snapshot", + object = %DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + "observed data usage snapshot was not a structurally complete nonconverged view" + ); + None + } + Err(err) => { + record_usage_snapshot_decode_failure("parse_observed", DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), &err); + None + } + } +} + +fn select_admin_data_usage_snapshot( + mut authoritative: DataUsageInfo, + authoritative_format: bool, + observed: Option, +) -> (DataUsageInfo, bool) { + if authoritative_format + && authoritative.is_complete_bucket_usage_snapshot() + && authoritative.usage_snapshot_converged.is_none() + { + authoritative.usage_snapshot_converged = Some(true); + } + match observed { + Some(observed) if observed_data_usage_is_newer(&observed, &authoritative) => (observed, true), + _ => (authoritative, authoritative_format), + } +} + +async fn load_admin_data_usage_from_backend(store: Arc) -> Result { + let (authoritative, source) = load_data_usage_snapshot(store.clone()).await?; + let observed = load_observed_data_usage_snapshot(store).await; + let (selected, selected_is_current_format) = + select_admin_data_usage_snapshot(authoritative, source.is_authoritative(), observed); + Ok(normalize_loaded_data_usage(selected, selected_is_current_format).await.0) +} + fn discard_incomplete_bucket_usage(data_usage_info: &mut DataUsageInfo) { if !data_usage_info.is_complete_bucket_usage_snapshot() { data_usage_info.usage_snapshot_complete = false; @@ -944,7 +1113,55 @@ pub async fn load_data_usage_from_backend_cached(store: Arc) -> Result< let result = load_data_usage_from_backend_with_baseline(store.clone()).await; let loaded_at = tokio::time::Instant::now(); let mut cache = data_usage_snapshot_cache().write().await; - if let Some(result) = cache_data_usage_snapshot_result(&mut cache, result, loaded_at, refresh_generation) { + if let Some(result) = + cache_data_usage_snapshot_result(&mut cache, result, loaded_at, refresh_generation, data_usage_snapshot_generation()) + { + return result; + } + drop(cache); + drop(refresh_guard); + } +} + +/// Load the freshest structurally complete snapshot for authenticated admin +/// observability. A scan raced by namespace activity may be selected here, but +/// never by [`load_data_usage_from_backend_cached`], which remains the +/// converged source for quota admission. +pub async fn load_admin_data_usage_from_backend_cached(store: Arc) -> Result { + let ttl = Duration::from_secs(DATA_USAGE_CACHE_TTL_SECS); + + loop { + { + let cache = admin_data_usage_snapshot_cache().read().await; + if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl) { + return result; + } + } + + let refresh_guard = ADMIN_DATA_USAGE_SNAPSHOT_REFRESH + .get_or_init(|| Arc::new(TokioMutex::new(()))) + .lock() + .await; + { + let cache = admin_data_usage_snapshot_cache().read().await; + if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl) { + return result; + } + } + + let refresh_generation = admin_data_usage_snapshot_generation(); + let result = load_admin_data_usage_from_backend(store.clone()) + .await + .map(|info| (info, HashMap::new())); + let loaded_at = tokio::time::Instant::now(); + let mut cache = admin_data_usage_snapshot_cache().write().await; + if let Some(result) = cache_data_usage_snapshot_result( + &mut cache, + result, + loaded_at, + refresh_generation, + admin_data_usage_snapshot_generation(), + ) { return result; } drop(cache); @@ -957,6 +1174,16 @@ pub async fn load_data_usage_from_backend_cached(store: Arc) -> Result< pub async fn invalidate_data_usage_snapshot_cache() { let mut cache = data_usage_snapshot_cache().write().await; clear_data_usage_snapshot_cache(&mut cache); + + let mut admin_cache = admin_data_usage_snapshot_cache().write().await; + clear_admin_data_usage_snapshot_cache(&mut admin_cache); +} + +/// Invalidate only the admin/console view after an observational save. Quota +/// admission continues to use the independently cached converged snapshot. +pub async fn invalidate_admin_data_usage_snapshot_cache() { + let mut cache = admin_data_usage_snapshot_cache().write().await; + clear_admin_data_usage_snapshot_cache(&mut cache); } /// Aggregate usage information from local disk snapshots. @@ -2080,6 +2307,7 @@ mod tests { struct UsageCasState { object: Option<(Vec, u64)>, backup_object: Option<(Vec, u64)>, + observed_object: Option<(Vec, u64)>, legacy_object: Option<(Vec, u64)>, legacy_backup_object: Option<(Vec, u64)>, interleaving_snapshot: Option>, @@ -2099,10 +2327,24 @@ mod tests { state: Mutex, } + #[async_trait::async_trait] + impl ObservedDataUsageSnapshotCleanup for UsageCasStore { + async fn delete_observed_data_usage_snapshot(&self, revision: &str) -> Result<(), Error> { + let mut state = self.state.lock().await; + let current = state.observed_object.as_ref().ok_or(Error::FileNotFound)?.1; + if revision != format!("usage-{current}") { + return Err(Error::PreconditionFailed); + } + state.observed_object = None; + Ok(()) + } + } + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum UsageObjectSlot { Primary, Backup, + Observed, LegacyPrimary, LegacyBackup, } @@ -2131,6 +2373,7 @@ mod tests { let slot = match object { object if object == DATA_USAGE_OBJ_NAME_PATH.as_str() => UsageObjectSlot::Primary, object if object == DATA_USAGE_OBJ_BACKUP_PATH.as_str() => UsageObjectSlot::Backup, + object if object == DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str() => UsageObjectSlot::Observed, object if object == LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str() => UsageObjectSlot::LegacyPrimary, object if object == LEGACY_DATA_USAGE_OBJ_BACKUP_PATH.as_str() => UsageObjectSlot::LegacyBackup, _ => return Err(Error::FileNotFound), @@ -2139,6 +2382,7 @@ mod tests { let stored = match slot { UsageObjectSlot::Primary => &state.object, UsageObjectSlot::Backup => &state.backup_object, + UsageObjectSlot::Observed => &state.observed_object, UsageObjectSlot::LegacyPrimary => &state.legacy_object, UsageObjectSlot::LegacyBackup => &state.legacy_backup_object, }; @@ -2178,6 +2422,7 @@ mod tests { let slot = match object { object if object == DATA_USAGE_OBJ_NAME_PATH.as_str() => UsageObjectSlot::Primary, object if object == DATA_USAGE_OBJ_BACKUP_PATH.as_str() => UsageObjectSlot::Backup, + object if object == DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str() => UsageObjectSlot::Observed, object if object == LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str() => UsageObjectSlot::LegacyPrimary, object if object == LEGACY_DATA_USAGE_OBJ_BACKUP_PATH.as_str() => UsageObjectSlot::LegacyBackup, _ => return Err(Error::FileNotFound), @@ -2213,6 +2458,9 @@ mod tests { let revision = state.backup_object.as_ref().map_or(1, |(_, revision)| revision + 1); state.backup_object = Some((interleaving, revision)); } + if slot == UsageObjectSlot::Observed { + return Err(Error::other("observed test fixture writes are injected directly")); + } if slot == UsageObjectSlot::LegacyPrimary && let Some(interleaving) = state.interleaving_legacy_snapshot.take() { @@ -2228,6 +2476,7 @@ mod tests { let current_revision = match slot { UsageObjectSlot::Primary => state.object.as_ref(), UsageObjectSlot::Backup => state.backup_object.as_ref(), + UsageObjectSlot::Observed => state.observed_object.as_ref(), UsageObjectSlot::LegacyPrimary => state.legacy_object.as_ref(), UsageObjectSlot::LegacyBackup => state.legacy_backup_object.as_ref(), } @@ -2258,6 +2507,7 @@ mod tests { match slot { UsageObjectSlot::Primary => state.object = Some((buf, revision)), UsageObjectSlot::Backup => state.backup_object = Some((buf, revision)), + UsageObjectSlot::Observed => state.observed_object = Some((buf, revision)), UsageObjectSlot::LegacyPrimary => state.legacy_object = Some((buf, revision)), UsageObjectSlot::LegacyBackup => state.legacy_backup_object = Some((buf, revision)), } @@ -2550,6 +2800,71 @@ mod tests { assert!(normalized.buckets_usage.is_empty()); } + #[test] + fn admin_snapshot_selection_requires_the_current_authoritative_baseline() { + let authoritative = DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH), + scanner_epoch: Some(4), + scanner_cycle: Some(10), + usage_snapshot_complete: true, + ..Default::default() + }; + let observed = DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1)), + scanner_epoch: Some(4), + scanner_cycle: Some(11), + usage_snapshot_complete: true, + usage_snapshot_converged: Some(false), + usage_snapshot_authoritative_baseline: Some(authoritative.snapshot_identity()), + ..Default::default() + }; + + let (selected, _) = select_admin_data_usage_snapshot(authoritative.clone(), true, Some(observed.clone())); + assert_eq!(selected.usage_snapshot_converged, Some(false)); + + let mut namespace_changed = authoritative; + namespace_changed.last_update = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(2)); + let (selected, _) = select_admin_data_usage_snapshot(namespace_changed, true, Some(observed)); + assert_eq!(selected.usage_snapshot_converged, Some(true)); + } + + #[tokio::test] + async fn authoritative_save_cleanup_removes_observed_snapshot_best_effort() { + let store = UsageCasStore::default(); + let authoritative = data_usage_info_for_test("bucket", 1, 10, SystemTime::UNIX_EPOCH + Duration::from_secs(2)); + let stale_observed = DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1)), + scanner_epoch: Some(4), + scanner_cycle: Some(10), + usage_snapshot_complete: true, + usage_snapshot_converged: Some(false), + ..Default::default() + }; + store.state.lock().await.observed_object = + Some((serde_json::to_vec(&stale_observed).expect("observed snapshot should encode"), 1)); + + cleanup_observed_data_usage_after_authoritative_save(&store, &authoritative).await; + assert!(store.state.lock().await.observed_object.is_none()); + + cleanup_observed_data_usage_after_authoritative_save(&store, &authoritative).await; + assert!(store.state.lock().await.observed_object.is_none()); + + let newer_observed = DataUsageInfo { + last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(3)), + scanner_epoch: Some(4), + scanner_cycle: Some(11), + usage_snapshot_complete: true, + usage_snapshot_converged: Some(false), + usage_snapshot_authoritative_baseline: Some(authoritative.snapshot_identity()), + ..Default::default() + }; + store.state.lock().await.observed_object = + Some((serde_json::to_vec(&newer_observed).expect("observed snapshot should encode"), 2)); + + cleanup_observed_data_usage_after_authoritative_save(&store, &authoritative).await; + assert!(store.state.lock().await.observed_object.is_some()); + } + #[test] #[serial] fn cached_snapshot_failure_is_reused_until_ttl_expires() { @@ -2557,8 +2872,14 @@ mod tests { let mut cache = None; let refresh_generation = data_usage_snapshot_generation(); - let first = cache_data_usage_snapshot_result(&mut cache, Err(Error::ErasureReadQuorum), loaded_at, refresh_generation) - .expect("an uninterrupted refresh should populate the cache"); + let first = cache_data_usage_snapshot_result( + &mut cache, + Err(Error::ErasureReadQuorum), + loaded_at, + refresh_generation, + data_usage_snapshot_generation(), + ) + .expect("an uninterrupted refresh should populate the cache"); assert!(matches!(first, Err(Error::ErasureReadQuorum))); let cached = fresh_cached_data_usage_snapshot(&cache, loaded_at + Duration::from_secs(1), Duration::from_secs(30)) @@ -2576,9 +2897,15 @@ mod tests { let mut cache = None; let refresh_generation = data_usage_snapshot_generation(); - let first = cache_data_usage_snapshot_result(&mut cache, Ok((expected, HashMap::new())), loaded_at, refresh_generation) - .expect("an uninterrupted refresh should populate the cache") - .expect("successful load must be returned"); + let first = cache_data_usage_snapshot_result( + &mut cache, + Ok((expected, HashMap::new())), + loaded_at, + refresh_generation, + data_usage_snapshot_generation(), + ) + .expect("an uninterrupted refresh should populate the cache") + .expect("successful load must be returned"); assert_snapshot_bucket(&first, "bucket"); let cached = fresh_cached_data_usage_snapshot(&cache, loaded_at + Duration::from_secs(1), Duration::from_secs(30)) @@ -2604,6 +2931,7 @@ mod tests { Ok((data_usage_info_for_test("stale", 1, 42, SystemTime::UNIX_EPOCH), HashMap::new())), loaded_at, refresh_generation, + data_usage_snapshot_generation(), ); assert!(stale_result.is_none()); diff --git a/crates/madmin/src/user.rs b/crates/madmin/src/user.rs index b6c9ca249..e4d7ff4fd 100644 --- a/crates/madmin/src/user.rs +++ b/crates/madmin/src/user.rs @@ -401,6 +401,11 @@ pub struct AccountInfo { pub server: BackendInfo, pub policy: serde_json::Value, // Use iam/policy::parse to parse the result, to be done by the caller. pub buckets: Vec, + /// Whether the usage values came from a fully converged scanner cycle. + /// `false` means the values are a newer structurally complete observation + /// while another convergence pass remains pending. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub usage_snapshot_converged: Option, } #[derive(Debug, Serialize, Deserialize, Default)] @@ -798,6 +803,20 @@ mod tests { assert_eq!(value["object_versions_histogram"], serde_json::json!({})); } + #[test] + fn account_info_exposes_observational_usage_state_additively() { + let info = AccountInfo { + usage_snapshot_converged: Some(false), + ..Default::default() + }; + + let value = serde_json::to_value(&info).unwrap(); + assert_eq!(value["usage_snapshot_converged"], false); + + let legacy_shape = serde_json::to_value(AccountInfo::default()).unwrap(); + assert!(!legacy_shape.as_object().unwrap().contains_key("usage_snapshot_converged")); + } + #[test] fn test_account_status_try_from_invalid() { let result = AccountStatus::try_from("invalid"); diff --git a/crates/obs/src/metrics/collectors/cluster_usage.rs b/crates/obs/src/metrics/collectors/cluster_usage.rs index 85e6cb1d7..c493145ec 100644 --- a/crates/obs/src/metrics/collectors/cluster_usage.rs +++ b/crates/obs/src/metrics/collectors/cluster_usage.rs @@ -38,6 +38,9 @@ pub struct ClusterUsageStats { pub delete_markers_count: u64, /// Total number of buckets in the usage snapshot pub buckets_count: u64, + /// Whether the selected admin usage snapshot completed without concurrent + /// namespace activity. + pub snapshot_converged: bool, /// Object size distribution by range pub object_size_distribution: Vec<(String, u64)>, /// Version count distribution by range @@ -69,7 +72,7 @@ pub struct BucketUsageStats { /// /// Returns a vector of Prometheus metrics for cluster usage. pub fn collect_cluster_usage_metrics(stats: &ClusterUsageStats) -> Vec { - let mut metrics = Vec::with_capacity(6 + stats.object_size_distribution.len() + stats.versions_distribution.len()); + let mut metrics = Vec::with_capacity(7 + stats.object_size_distribution.len() + stats.versions_distribution.len()); metrics.push(PrometheusMetric::from_descriptor( &USAGE_SINCE_LAST_UPDATE_SECONDS_MD, @@ -83,6 +86,10 @@ pub fn collect_cluster_usage_metrics(stats: &ClusterUsageStats) -> Vec = LazyLock::new(|| ) }); +pub static USAGE_SNAPSHOT_CONVERGED_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::UsageSnapshotConverged, + "Whether the selected admin usage snapshot is converged (1) or observational (0)", + &[], + subsystems::CLUSTER_USAGE_OBJECTS, + ) +}); + pub static USAGE_OBJECTS_DISTRIBUTION_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::UsageSizeDistribution, diff --git a/crates/obs/src/metrics/schema/entry/metric_name.rs b/crates/obs/src/metrics/schema/entry/metric_name.rs index a240acb03..7d22eaa62 100644 --- a/crates/obs/src/metrics/schema/entry/metric_name.rs +++ b/crates/obs/src/metrics/schema/entry/metric_name.rs @@ -233,6 +233,7 @@ pub enum MetricName { UsageVersionsCount, UsageDeleteMarkersCount, UsageBucketsCount, + UsageSnapshotConverged, UsageSizeDistribution, UsageVersionCountDistribution, @@ -642,6 +643,7 @@ impl MetricName { Self::UsageVersionsCount => "versions_count".to_string(), Self::UsageDeleteMarkersCount => "delete_markers_count".to_string(), Self::UsageBucketsCount => "buckets_count".to_string(), + Self::UsageSnapshotConverged => "snapshot_converged".to_string(), Self::UsageSizeDistribution => "size_distribution".to_string(), Self::UsageVersionCountDistribution => "version_count_distribution".to_string(), diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 9e9ea26f8..c312f7eff 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -65,6 +65,7 @@ type ObsBackendInfo = ::BackendInfo; struct ObsDataUsageInfo { last_update: Option, usage_snapshot_complete: bool, + usage_snapshot_converged: bool, buckets_count: u64, objects_total_count: u64, versions_total_count: u64, @@ -102,6 +103,7 @@ async fn load_obs_data_usage_from_backend(store: Arc) -> ObsEcstoreRes Ok(ObsDataUsageInfo { last_update: data_usage.last_update, usage_snapshot_complete, + usage_snapshot_converged: data_usage.usage_snapshot_converged == Some(true), buckets_count: data_usage.buckets_count, objects_total_count: data_usage.objects_total_count, versions_total_count: data_usage.versions_total_count, @@ -1204,6 +1206,7 @@ async fn collect_cluster_usage_metric_stats_from_data_usage( versions_count: data_usage.versions_total_count, delete_markers_count: data_usage.delete_markers_total_count, buckets_count: data_usage.buckets_count, + snapshot_converged: data_usage.usage_snapshot_converged, object_size_distribution: data_usage .buckets_usage .values() @@ -1579,6 +1582,7 @@ mod tests { let (cluster, buckets) = collect_cluster_usage_metric_stats_from_data_usage( ObsDataUsageInfo { usage_snapshot_complete: true, + usage_snapshot_converged: true, ..Default::default() }, &HashSet::new(), @@ -1589,9 +1593,26 @@ mod tests { assert_eq!(cluster.buckets_count, 0); assert_eq!(cluster.objects_count, 0); assert_eq!(cluster.total_bytes, 0); + assert!(cluster.snapshot_converged); assert!(buckets.is_empty()); } + #[tokio::test] + async fn cluster_usage_metrics_publish_unknown_convergence_as_unconverged() { + let (cluster, _) = collect_cluster_usage_metric_stats_from_data_usage( + ObsDataUsageInfo { + usage_snapshot_complete: true, + usage_snapshot_converged: false, + ..Default::default() + }, + &HashSet::new(), + ) + .await + .expect("complete usage with unknown convergence should remain publishable"); + + assert!(!cluster.snapshot_converged); + } + #[tokio::test] async fn cluster_usage_metrics_skip_snapshot_for_a_different_bucket_namespace() { let data_usage = ObsDataUsageInfo { diff --git a/crates/obs/src/metrics/storage_api.rs b/crates/obs/src/metrics/storage_api.rs index 452471632..073bfb488 100644 --- a/crates/obs/src/metrics/storage_api.rs +++ b/crates/obs/src/metrics/storage_api.rs @@ -27,8 +27,8 @@ pub(crate) use rustfs_ecstore::api::capacity::{ get_total_usable_capacity_free as obs_get_total_usable_capacity_free, }; pub(crate) use rustfs_ecstore::api::compression::is_disk_compression_enabled as obs_is_disk_compression_enabled; +pub(crate) use rustfs_ecstore::api::data_usage::load_admin_data_usage_from_backend_cached as obs_load_data_usage_from_backend; pub(crate) use rustfs_ecstore::api::data_usage::load_compression_total_from_memory as obs_load_compression_total_from_memory; -pub(crate) use rustfs_ecstore::api::data_usage::load_data_usage_from_backend as obs_load_data_usage_from_backend; pub(crate) use rustfs_ecstore::api::error::Result as ObsEcstoreResult; pub(crate) use rustfs_ecstore::api::runtime::{ bucket_monitor as obs_get_global_bucket_monitor, expiry_state_handle as obs_expiry_state_handle, diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 5c36c59ce..ff827fcc5 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -27,8 +27,8 @@ use rustfs_common::heal_channel::HealScanMode; #[cfg(test)] use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS; pub use rustfs_data_usage::{ - AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DataUsageEntry, DataUsageHash, - DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, TierStats, hash_path, + AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME, + DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, TierStats, hash_path, }; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; use tokio::time::{Duration, Instant, sleep, timeout}; @@ -173,6 +173,9 @@ pub static DATA_USAGE_BUCKET: LazyLock = pub static DATA_USAGE_OBJ_NAME_PATH: LazyLock = LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{DATA_USAGE_OBJECT_NAME}")); +pub static DATA_USAGE_OBSERVED_OBJ_NAME_PATH: LazyLock = + LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{DATA_USAGE_OBSERVED_OBJECT_NAME}")); + pub static LEGACY_DATA_USAGE_OBJ_NAME_PATH: LazyLock = LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{LEGACY_DATA_USAGE_OBJECT_NAME}")); diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 3341bba2d..02758948d 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -35,10 +35,11 @@ use storage_api::owner::{ EcstoreTierConfig, EcstoreVersioningApi, HTTPPreconditions, HTTPRangeSpec, ObjectIO, ObjectOperations, ObjectToDelete, ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, - ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_data_usage_snapshot_cache, - ecstore_is_erasure, ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, - ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, - ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config, + ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache, + ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_erasure_sd, + ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info, + ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_read_config, + ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config, scanner_replication_config_for_lifecycle_eval, }; #[cfg(test)] @@ -475,6 +476,10 @@ pub(crate) async fn invalidate_data_usage_snapshot_cache() { ecstore_invalidate_data_usage_snapshot_cache().await; } +pub(crate) async fn invalidate_admin_data_usage_snapshot_cache() { + ecstore_invalidate_admin_data_usage_snapshot_cache().await; +} + pub trait ScannerObjectIO: ObjectIO< Error = EcstoreError, @@ -501,6 +506,28 @@ impl ScannerObjectIO for T where { } +#[async_trait::async_trait] +pub trait ScannerConfigObjectDelete: Send + Sync + std::fmt::Debug + 'static { + async fn delete_config_object( + &self, + bucket: &str, + object: &str, + opts: ScannerObjectOptions, + ) -> EcstoreResult; +} + +#[async_trait::async_trait] +impl ScannerConfigObjectDelete for ECStore { + async fn delete_config_object( + &self, + bucket: &str, + object: &str, + opts: ScannerObjectOptions, + ) -> EcstoreResult { + ObjectOperations::delete_object(self, bucket, object, opts).await + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 6d10fbf24..f8229b57b 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -18,10 +18,9 @@ use std::future::Future; use std::sync::Mutex as StdMutex; use std::sync::{Arc, LazyLock, RwLock}; -use crate::ScannerObjectIO; use crate::data_usage_define::{ - BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH, DataUsageCache, DataUsageCacheRevision, - LEGACY_DATA_USAGE_OBJ_NAME_PATH, read_config_with_revision, + BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH, DATA_USAGE_OBSERVED_OBJ_NAME_PATH, + DataUsageCache, DataUsageCacheRevision, LEGACY_DATA_USAGE_OBJ_NAME_PATH, read_config_with_revision, }; use crate::runtime_config::{ ScannerRuntimeConfig, ScannerRuntimeConfigSource, refresh_scanner_runtime_config_from_global, scanner_bitrot_cycle, @@ -36,6 +35,7 @@ use crate::scanner_io::{ }; use crate::sleeper::{SCANNER_SLEEPER, set_scanner_default_speed}; use crate::{DataUsageInfo, ScannerActivityGuard, ScannerError}; +use crate::{ScannerConfigObjectDelete, ScannerObjectIO, ScannerObjectOptions}; use bytes::Bytes; use chrono::{DateTime, Utc}; use rustfs_common::heal_channel::HealScanMode; @@ -50,6 +50,7 @@ use rustfs_config::{ ENV_SCANNER_CYCLE_MAX_OBJECTS, }; use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS}; +use rustfs_data_usage::observed_data_usage_is_newer; use serde::{Deserialize, Serialize}; use sha2::{Digest as _, Sha256}; #[cfg(test)] @@ -66,9 +67,9 @@ use crate::storage_api::scan::{ }; use crate::{ ECStore, EcstoreError, RUSTFS_META_BUCKET, ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _, - get_lifecycle_config, get_replication_config, invalidate_data_usage_snapshot_cache, read_config, - replace_bucket_usage_memory_from_info, save_config, save_config_shared_with_preconditions, save_config_with_preconditions, - scanner_is_erasure_sd, + get_lifecycle_config, get_replication_config, invalidate_admin_data_usage_snapshot_cache, + invalidate_data_usage_snapshot_cache, read_config, replace_bucket_usage_memory_from_info, save_config, + save_config_shared_with_preconditions, save_config_with_preconditions, scanner_is_erasure_sd, }; const LOG_COMPONENT_SCANNER: &str = "scanner"; @@ -83,6 +84,8 @@ const METRIC_SCANNER_LEADER_LOCK_TOTAL: &str = "rustfs_scanner_leader_lock_total const CLEAN_IDLE_MAX_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60); const MAX_SCANNER_SCHEDULE_DELAY: Duration = Duration::from_secs(365 * 24 * 60 * 60); const CLEAN_IDLE_BACKOFF_FACTOR: u32 = 2; +const SUPERSEDED_RETRY_BASE_INTERVAL: Duration = Duration::from_secs(60); +const SUPERSEDED_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(30 * 60); const SCANNER_LEADER_LOCK_POLL_INTERVAL: Duration = Duration::from_secs(1); #[cfg(not(test))] const SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30); @@ -159,6 +162,8 @@ pub struct ScannerCycleScheduleStatus { effective_interval_seconds: u64, clean_idle_backoff_enabled: bool, clean_idle_backoff_multiplier: u64, + superseded_retry_backoff_enabled: bool, + superseded_cycles: u32, } impl Default for ScannerCycleScheduleStatus { @@ -167,6 +172,8 @@ impl Default for ScannerCycleScheduleStatus { effective_interval_seconds: 0, clean_idle_backoff_enabled: false, clean_idle_backoff_multiplier: 1, + superseded_retry_backoff_enabled: false, + superseded_cycles: 0, } } } @@ -188,6 +195,8 @@ fn record_scanner_cycle_schedule( effective_interval: Duration, clean_idle_backoff_enabled: bool, clean_idle_backoff_multiplier: u64, + superseded_retry_backoff_enabled: bool, + superseded_cycles: u32, ) { let effective_interval_seconds = effective_interval .as_secs() @@ -199,11 +208,13 @@ fn record_scanner_cycle_schedule( effective_interval_seconds, clean_idle_backoff_enabled, clean_idle_backoff_multiplier: clean_idle_backoff_multiplier.max(1), + superseded_retry_backoff_enabled, + superseded_cycles, }; } fn reset_scanner_cycle_schedule() { - record_scanner_cycle_schedule(Duration::ZERO, false, 1); + record_scanner_cycle_schedule(Duration::ZERO, false, 1, false, 0); } /// Returns the base cycle interval. @@ -354,6 +365,37 @@ struct ScannerCleanIdleBackoff { interval_multiplier: u32, } +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +struct ScannerSupersededBackoff { + consecutive_cycles: u32, +} + +impl ScannerSupersededBackoff { + fn record_cycle(&mut self, outcome: ScannerCycleOutcome) { + match outcome { + ScannerCycleOutcome::Superseded => { + self.consecutive_cycles = self.consecutive_cycles.saturating_add(1); + } + ScannerCycleOutcome::Completed + | ScannerCycleOutcome::CompletedWithPendingMaintenance + | ScannerCycleOutcome::Partial + | ScannerCycleOutcome::Failed => { + self.consecutive_cycles = 0; + } + } + } + + fn retry_interval(self, configured_interval: Duration) -> Option { + let exponent = self.consecutive_cycles.checked_sub(1)?.min(31); + let multiplier = 1u32.checked_shl(exponent).unwrap_or(u32::MAX); + let base_interval = configured_interval + .max(Duration::from_secs(1)) + .min(SUPERSEDED_RETRY_BASE_INTERVAL); + let cap = SUPERSEDED_RETRY_MAX_INTERVAL.max(configured_interval.max(Duration::from_secs(1))); + Some(base_interval.saturating_mul(multiplier).min(cap)) + } +} + impl Default for ScannerCleanIdleBackoff { fn default() -> Self { Self { interval_multiplier: 1 } @@ -466,9 +508,10 @@ struct ScannerCycleWaitPlan { #[derive(Clone, Copy, Debug, PartialEq, Eq)] struct ScannerCycleObservedGenerations { - dirty_usage: u64, + dirty_usage: Option, runtime_config: u64, maintenance: u64, + defer_cluster_activity: bool, } const LOCAL_SCANNER_ACTIVITY_NODE: &str = ""; @@ -610,7 +653,7 @@ fn scanner_activity_backoff_blocked_after_wake(currently_blocked: bool, wake_rea async fn wait_for_next_scanner_cycle( ctx: &CancellationToken, delay: Duration, - dirty_usage_generation_seen: u64, + dirty_usage_generation_seen: Option, runtime_config_generation: u64, maintenance_generation: u64, is_lock_lost: F, @@ -633,7 +676,7 @@ where if scanner_maintenance_generation() != maintenance_generation { return ScannerCycleWakeReason::MaintenanceConfig; } - if dirty_usage_buckets_pending() && dirty_usage_generation() != dirty_usage_generation_seen { + if dirty_usage_generation_seen.is_some_and(|seen| dirty_usage_buckets_pending() && dirty_usage_generation() != seen) { return ScannerCycleWakeReason::DirtyUsage; } @@ -653,7 +696,9 @@ where if scanner_maintenance_generation() != maintenance_generation { return ScannerCycleWakeReason::MaintenanceConfig; } - if dirty_usage_buckets_pending() && dirty_usage_generation() != dirty_usage_generation_seen { + if dirty_usage_generation_seen + .is_some_and(|seen| dirty_usage_buckets_pending() && dirty_usage_generation() != seen) + { return ScannerCycleWakeReason::DirtyUsage; } } @@ -738,9 +783,15 @@ where } match observation { ScannerActivityObservation::Unchanged | ScannerActivityObservation::NotRequired => {} - ScannerActivityObservation::Changed => return ScannerCycleWakeReason::ClusterActivity, + ScannerActivityObservation::Changed if !generations.defer_cluster_activity => { + return ScannerCycleWakeReason::ClusterActivity; + } + ScannerActivityObservation::Changed => {} ScannerActivityObservation::MaintenanceChanged => return ScannerCycleWakeReason::ClusterMaintenance, - ScannerActivityObservation::Unverified => return ScannerCycleWakeReason::ClusterActivityUnavailable, + ScannerActivityObservation::Unverified if !generations.defer_cluster_activity => { + return ScannerCycleWakeReason::ClusterActivityUnavailable; + } + ScannerActivityObservation::Unverified => {} } } } @@ -1032,6 +1083,11 @@ fn data_usage_info_is_cold(info: &DataUsageInfo) -> bool { !info.is_complete_bucket_usage_snapshot() } +fn usage_cache_needs_prompt_scan(authoritative: &DataUsageInfo, observed: Option<&DataUsageInfo>) -> bool { + data_usage_info_is_cold(authoritative) + || observed.is_some_and(|observed| observed_data_usage_is_newer(observed, authoritative)) +} + async fn read_data_usage_config_for_startup(storeapi: &Arc) -> Result>, EcstoreError> { async fn read_pair(storeapi: &Arc, primary_path: &str) -> Result>, EcstoreError> { match read_config(storeapi.clone(), primary_path).await { @@ -1142,7 +1198,43 @@ async fn persisted_usage_cache_is_cold_for_startup(storeapi: &Arc) -> b }; match serde_json::from_slice::(&data) { - Ok(info) => data_usage_info_is_cold(&info), + Ok(info) => { + if data_usage_info_is_cold(&info) { + return true; + } + match read_config(storeapi.clone(), DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()).await { + Ok(observed) => match serde_json::from_slice::(&observed) { + Ok(observed) => usage_cache_needs_prompt_scan(&info, Some(&observed)), + Err(err) => { + warn!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + state = "startup_observed_decode_failed", + error = %err, + "Scanner startup found an invalid observational snapshot and will refresh it promptly" + ); + true + } + }, + Err(EcstoreError::ConfigNotFound) => false, + Err(err) => { + warn!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + state = "startup_observed_inspect_failed", + error = %err, + "Scanner startup could not inspect the observational snapshot" + ); + false + } + } + } Err(err) => { warn!( target: "rustfs::scanner", @@ -1853,6 +1945,17 @@ fn advance_scanner_cycle(cycle_info: &mut CurrentCycle) -> Result<(), ScannerCyc async fn persisted_usage_floor(storeapi: Arc) -> Result { let mut floor = PersistedUsageFloor::default(); + let update_floor = |floor: &mut PersistedUsageFloor, usage: DataUsageInfo, path: &str| -> Result<(), ScannerError> { + floor.leader_epoch = floor.leader_epoch.max(usage.scanner_epoch.unwrap_or_default()); + if let Some(completed_cycle) = usage.scanner_cycle { + let next_cycle = completed_cycle + .checked_add(1) + .filter(|next| *next < u64::MAX) + .ok_or_else(|| ScannerError::Other(format!("persisted scanner usage cycle is exhausted in {path}")))?; + floor.next_cycle = floor.next_cycle.max(next_cycle); + } + Ok(()) + }; for primary_path in [DATA_USAGE_OBJ_NAME_PATH.as_str(), LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()] { let backup_path = format!("{primary_path}.bkp"); let mut pair_found = false; @@ -1871,14 +1974,7 @@ async fn persisted_usage_floor(storeapi: Arc) -> Result(&data) .map_err(|err| ScannerError::Other(format!("failed to decode scanner usage floor from {path}: {err}")))?; - floor.leader_epoch = floor.leader_epoch.max(usage.scanner_epoch.unwrap_or_default()); - if let Some(completed_cycle) = usage.scanner_cycle { - let next_cycle = completed_cycle - .checked_add(1) - .filter(|next| *next < u64::MAX) - .ok_or_else(|| ScannerError::Other(format!("persisted scanner usage cycle is exhausted in {path}")))?; - floor.next_cycle = floor.next_cycle.max(next_cycle); - } + update_floor(&mut floor, usage, path)?; } if pair_found { break; @@ -3087,6 +3183,7 @@ async fn run_data_scanner_with_maintenance_state( let mut dirty_usage_generation_seen = dirty_usage_generation(); let mut runtime_config_generation_seen = scanner_runtime_config_generation(); let mut clean_idle_backoff = ScannerCleanIdleBackoff::default(); + let mut superseded_backoff = ScannerSupersededBackoff::default(); let initial_runtime_config = resolve_scanner_runtime_config(); if clean_idle_topology_supported && scanner_clean_idle_backoff_configured(&initial_runtime_config) @@ -3216,6 +3313,7 @@ async fn run_data_scanner_with_maintenance_state( ) .await .unwrap_or(ScannerCycleOutcome::Failed); + superseded_backoff.record_cycle(initial_outcome); dirty_usage_generation_seen = dirty_generation_before_cycle; if guard.is_lock_lost() { record_scanner_leader_lock_lost("Scanner leader lock lost during the initial cycle").await; @@ -3297,7 +3395,13 @@ async fn run_data_scanner_with_maintenance_state( maintenance_features, &runtime_config, ); - let wait_plan = scanner_cycle_wait_plan(&runtime_config, clean_idle_backoff, backoff_enabled, randomized_cycle_delay_for); + let mut wait_plan = + scanner_cycle_wait_plan(&runtime_config, clean_idle_backoff, backoff_enabled, randomized_cycle_delay_for); + let superseded_retry_interval = superseded_backoff.retry_interval(runtime_config.cycle_interval); + if let Some(retry_interval) = superseded_retry_interval { + wait_plan.effective_interval = retry_interval; + wait_plan.delay = randomized_cycle_delay_for(retry_interval).min(retry_interval); + } let dirty_generation_before_wait = dirty_usage_generation(); let dirty_usage_pending_before_wait = dirty_usage_buckets_pending(); let maintenance_generation_before_wait = scanner_maintenance_generation(); @@ -3305,6 +3409,8 @@ async fn run_data_scanner_with_maintenance_state( wait_plan.effective_interval, backoff_enabled, u64::from(clean_idle_backoff.interval_multiplier), + superseded_retry_interval.is_some(), + superseded_backoff.consecutive_cycles, ); debug!( target: "rustfs::scanner", @@ -3317,6 +3423,8 @@ async fn run_data_scanner_with_maintenance_state( scheduled_delay = ?wait_plan.delay, interval_multiplier = clean_idle_backoff.interval_multiplier, clean_idle_backoff_enabled = backoff_enabled, + superseded_retry_backoff_enabled = superseded_retry_interval.is_some(), + superseded_cycles = superseded_backoff.consecutive_cycles, lifecycle_active = maintenance_features.lifecycle, replication_active = maintenance_features.replication, feature_inspection_failed = maintenance_features.inspection_failed, @@ -3331,9 +3439,13 @@ async fn run_data_scanner_with_maintenance_state( activity_poll_interval, &mut scanner_activity_seen, ScannerCycleObservedGenerations { - dirty_usage: dirty_usage_generation_seen, + // A superseded cycle already observed concurrent writes. Hold + // further dirty notifications until the bounded retry timer so + // a hot bucket cannot drive an unbroken full-scan loop. + dirty_usage: superseded_retry_interval.is_none().then_some(dirty_usage_generation_seen), runtime_config: runtime_config_generation_seen, maintenance: maintenance_generation_before_wait, + defer_cluster_activity: superseded_retry_interval.is_some(), }, || guard.is_lock_lost(), || probe_scanner_activity(storeapi.as_ref(), distributed), @@ -3410,6 +3522,7 @@ async fn run_data_scanner_with_maintenance_state( ) .await .unwrap_or(ScannerCycleOutcome::Failed); + superseded_backoff.record_cycle(outcome); dirty_usage_generation_seen = dirty_generation_before_cycle; if guard.is_lock_lost() { record_scanner_leader_lock_lost("Scanner leader lock lost during a scanner cycle").await; @@ -3606,10 +3719,12 @@ fn finalize_scanner_cycle_result( scan_cycle_result.has_failed_dirty_usage(), ); let pending_maintenance_work = scan_cycle_result.has_pending_maintenance_work(); - let remote_dirty_usage_acknowledgements = if matches!( - usage_persist_outcome, - DataUsagePersistOutcome::Saved | DataUsagePersistOutcome::AlreadyDurable - ) { + let durable_complete_snapshot = scan_cycle_result.status == ScannerCycleStatus::Complete + && matches!( + usage_persist_outcome, + DataUsagePersistOutcome::Saved | DataUsagePersistOutcome::AlreadyDurable + ); + let remote_dirty_usage_acknowledgements = if durable_complete_snapshot { scan_cycle_result.acknowledge_durable_usage() } else { Vec::new() @@ -3678,7 +3793,7 @@ fn data_usage_reintroduces_missing_bucket(incoming: &DataUsageInfo, existing: Op #[instrument(skip(ctx, storeapi))] pub async fn store_data_usage_in_backend( ctx: CancellationToken, - storeapi: Arc, + storeapi: Arc, receiver: mpsc::Receiver, ) { let _ = store_data_usage_in_backend_with_outcome(ctx, storeapi, receiver).await; @@ -3686,7 +3801,7 @@ pub async fn store_data_usage_in_backend( async fn store_data_usage_in_backend_with_outcome( ctx: CancellationToken, - storeapi: Arc, + storeapi: Arc, receiver: mpsc::Receiver, ) -> DataUsagePersistOutcome { store_data_usage_in_backend_with_outcome_for_epoch(ctx, storeapi, receiver, None).await @@ -3694,7 +3809,7 @@ async fn store_data_usage_in_backend_with_outcome( async fn store_data_usage_in_backend_with_outcome_for_epoch( ctx: CancellationToken, - storeapi: Arc, + storeapi: Arc, receiver: mpsc::Receiver, leader_epoch: Option, ) -> DataUsagePersistOutcome { @@ -3703,7 +3818,7 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch( async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( ctx: CancellationToken, - storeapi: Arc, + storeapi: Arc, mut receiver: mpsc::Receiver, leader_epoch: Option, initial_baseline: Option, @@ -3719,6 +3834,56 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( if let Some(leader_epoch) = leader_epoch { data_usage_info.scanner_epoch = Some(leader_epoch); } + let observational = data_usage_info.usage_snapshot_converged == Some(false); + let target_path = if observational { + DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str() + } else { + DATA_USAGE_OBJ_NAME_PATH.as_str() + }; + + if observational && data_usage_info.usage_snapshot_authoritative_baseline.is_none() { + let authoritative_data = match next_baseline.as_ref() { + Some(baseline) => baseline.data.clone(), + None => match read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await { + Ok((data, _)) => data.map(Bytes::from), + Err(err) => { + error!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), + state = "observed_baseline_load_failed", + error = %err, + "Scanner could not identify the authoritative baseline for an observation" + ); + outcome = DataUsagePersistOutcome::Failed; + continue; + } + }, + }; + let authoritative = match authoritative_data.as_deref() { + Some(data) => match serde_json::from_slice::(data) { + Ok(info) => info, + Err(err) => { + error!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), + state = "observed_baseline_decode_failed", + error = %err, + "Scanner refused to publish an observation from an invalid authoritative baseline" + ); + outcome = DataUsagePersistOutcome::Failed; + continue; + } + }, + None => DataUsageInfo::default(), + }; + data_usage_info.usage_snapshot_authoritative_baseline = Some(authoritative.snapshot_identity()); + } if !data_usage_info.is_complete_bucket_usage_snapshot() { error!( @@ -3726,7 +3891,7 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( event = EVENT_SCANNER_PERSIST_STATE, component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_RUNTIME, - path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), + path = %target_path, state = "reject_incomplete_snapshot", "Scanner refused to persist an incomplete data usage snapshot" ); @@ -3743,7 +3908,7 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( event = EVENT_SCANNER_PERSIST_STATE, component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_RUNTIME, - path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), + path = %target_path, state = "encode_failed", error = %e, "Scanner data usage encode failed" @@ -3755,17 +3920,21 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( }; let sha256hex = (!data.is_empty()).then(|| hex_simd::encode_to_string(Sha256::digest(&data), hex_simd::AsciiCase::Lower)); let data = Bytes::from(data); - let backup_due = data_usage_backup_due(&data_usage_info); + let backup_due = !observational && data_usage_backup_due(&data_usage_info); let mut cas_retry = 0usize; let save_outcome = loop { if ctx.is_cancelled() { break 'updates; } - let baseline = if cas_retry == 0 { next_baseline.take() } else { None }; + let baseline = if !observational && cas_retry == 0 { + next_baseline.take() + } else { + None + }; let (existing_data, revision) = match baseline { Some(baseline) => (baseline.data, baseline.revision), - None => match read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await { + None => match read_config_with_revision(storeapi.clone(), target_path).await { Ok((data, revision)) => (data.map(Bytes::from), revision), Err(e) => { error!( @@ -3773,7 +3942,7 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( event = EVENT_SCANNER_PERSIST_STATE, component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_RUNTIME, - path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), + path = %target_path, state = "revision_load_failed", error = %e, "Scanner data usage revision load failed" @@ -3791,7 +3960,7 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( event = EVENT_SCANNER_PERSIST_STATE, component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_RUNTIME, - path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), + path = %target_path, incoming_scanner_epoch = ?data_usage_info.scanner_epoch, incoming_scanner_cycle = ?data_usage_info.scanner_cycle, state = "skip_deleted_bucket_reintroduction", @@ -3816,7 +3985,7 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( event = EVENT_SCANNER_PERSIST_STATE, component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_RUNTIME, - path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), + path = %target_path, incoming_scanner_epoch = ?data_usage_info.scanner_epoch, existing_scanner_epoch = ?existing.scanner_epoch, incoming_scanner_cycle = ?data_usage_info.scanner_cycle, @@ -3837,7 +4006,7 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( let done_save = Metrics::time(Metric::SaveUsage); let save_result = save_config_shared_with_preconditions( storeapi.clone(), - DATA_USAGE_OBJ_NAME_PATH.as_str(), + target_path, data.clone(), sha256hex.clone(), revision.preconditions(), @@ -3847,13 +4016,15 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( match save_result { Ok(object_info) => { - next_baseline = object_info - .etag - .filter(|etag| !etag.is_empty()) - .map(|etag| DataUsagePersistBaseline { - data: Some(data.clone()), - revision: DataUsageCacheRevision::Etag(etag), - }); + if !observational { + next_baseline = object_info + .etag + .filter(|etag| !etag.is_empty()) + .map(|etag| DataUsagePersistBaseline { + data: Some(data.clone()), + revision: DataUsageCacheRevision::Etag(etag), + }); + } break DataUsagePersistOutcome::Saved; } Err(EcstoreError::PreconditionFailed) if cas_retry < SCANNER_PERSIST_CAS_RETRIES => { @@ -3863,7 +4034,7 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( event = EVENT_SCANNER_PERSIST_STATE, component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_RUNTIME, - path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), + path = %target_path, state = "conflict_retry", retry = cas_retry, "Scanner data usage CAS conflict will be reconciled" @@ -3875,7 +4046,7 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( event = EVENT_SCANNER_PERSIST_STATE, component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_RUNTIME, - path = %DATA_USAGE_OBJ_NAME_PATH.as_str(), + path = %target_path, state = if matches!(e, EcstoreError::PreconditionFailed) { "conflict_retries_exhausted" } else { @@ -3891,19 +4062,33 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( match save_outcome { DataUsagePersistOutcome::Current => { - invalidate_data_usage_snapshot_cache().await; + if observational { + invalidate_admin_data_usage_snapshot_cache().await; + } else { + invalidate_data_usage_snapshot_cache().await; + } global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::SkippedStale); outcome = DataUsagePersistOutcome::Current; continue; } DataUsagePersistOutcome::AlreadyDurable => { - invalidate_data_usage_snapshot_cache().await; - replace_bucket_usage_memory_from_info(&data_usage_info).await; + if observational { + invalidate_admin_data_usage_snapshot_cache().await; + } else { + cleanup_observed_data_usage_snapshot(storeapi.clone(), &data_usage_info).await; + invalidate_data_usage_snapshot_cache().await; + replace_bucket_usage_memory_from_info(&data_usage_info).await; + } global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::Success); outcome = DataUsagePersistOutcome::AlreadyDurable; } DataUsagePersistOutcome::PriorCycleDurable => { - invalidate_data_usage_snapshot_cache().await; + if observational { + invalidate_admin_data_usage_snapshot_cache().await; + } else { + cleanup_observed_data_usage_snapshot(storeapi.clone(), &data_usage_info).await; + invalidate_data_usage_snapshot_cache().await; + } global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::Success); outcome = DataUsagePersistOutcome::PriorCycleDurable; } @@ -3913,8 +4098,13 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( continue; } DataUsagePersistOutcome::Saved => { - invalidate_data_usage_snapshot_cache().await; - replace_bucket_usage_memory_from_info(&data_usage_info).await; + if observational { + invalidate_admin_data_usage_snapshot_cache().await; + } else { + cleanup_observed_data_usage_snapshot(storeapi.clone(), &data_usage_info).await; + invalidate_data_usage_snapshot_cache().await; + replace_bucket_usage_memory_from_info(&data_usage_info).await; + } global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::Success); outcome = DataUsagePersistOutcome::Saved; } @@ -3941,6 +4131,84 @@ async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline( outcome } +async fn cleanup_observed_data_usage_snapshot( + storeapi: Arc, + authoritative: &DataUsageInfo, +) { + let (observed_data, revision) = + match read_config_with_revision(storeapi.clone(), DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()).await { + Ok((Some(data), revision)) => (data, revision), + Ok((None, _)) => return, + Err(err) => { + error!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + state = "observed_cleanup_read_failed", + error = %err, + "Scanner could not inspect observational data usage snapshot before authoritative cleanup" + ); + return; + } + }; + let observed = match serde_json::from_slice::(&observed_data) { + Ok(observed) => observed, + Err(err) => { + error!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + state = "observed_cleanup_decode_failed", + error = %err, + "Scanner refused to remove an invalid observational data usage snapshot after authoritative save" + ); + return; + } + }; + if observed_data_usage_is_newer(&observed, authoritative) { + return; + } + + let result = storeapi + .delete_config_object( + RUSTFS_META_BUCKET, + DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + ScannerObjectOptions { + delete_prefix: true, + delete_prefix_object: true, + http_preconditions: Some(revision.preconditions()), + ..Default::default() + }, + ) + .await; + + match result { + Ok(_) + | Err( + EcstoreError::FileNotFound + | EcstoreError::ConfigNotFound + | EcstoreError::ObjectNotFound(_, _) + | EcstoreError::PreconditionFailed, + ) => {} + Err(err) => { + error!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + path = %DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(), + state = "observed_cleanup_failed", + error = %err, + "Scanner could not remove stale observational data usage snapshot after authoritative save" + ); + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -4823,6 +5091,35 @@ mod tests { })); } + #[test] + fn scanner_startup_prompts_only_for_a_newer_valid_observation() { + let authoritative = DataUsageInfo { + last_update: Some(std::time::SystemTime::UNIX_EPOCH), + scanner_epoch: Some(4), + scanner_cycle: Some(10), + ..complete_usage_with_bucket_count(None, 0) + }; + let observed = DataUsageInfo { + last_update: Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1)), + scanner_epoch: Some(4), + scanner_cycle: Some(11), + usage_snapshot_converged: Some(false), + usage_snapshot_authoritative_baseline: Some(authoritative.snapshot_identity()), + ..complete_usage_with_bucket_count(None, 0) + }; + + assert!(usage_cache_needs_prompt_scan(&authoritative, Some(&observed))); + assert!(!usage_cache_needs_prompt_scan(&authoritative, None)); + + let mut converged = observed.clone(); + converged.usage_snapshot_converged = Some(true); + assert!(!usage_cache_needs_prompt_scan(&authoritative, Some(&converged))); + + let mut legacy_observation = observed; + legacy_observation.usage_snapshot_converged = None; + assert!(!usage_cache_needs_prompt_scan(&authoritative, Some(&legacy_observation))); + } + #[tokio::test] async fn scanner_startup_prefers_v2_over_legacy_usage() { let store = Arc::new(MemoryConfigStore::default()); @@ -4958,6 +5255,31 @@ mod tests { } } + #[async_trait::async_trait] + impl crate::ScannerConfigObjectDelete for MemoryConfigStore { + async fn delete_config_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> EcstoreResult { + let key = memory_config_key(bucket, object); + let mut objects = self.objects.lock().await; + if !objects.contains_key(&key) { + return Err(EcstoreError::FileNotFound); + } + let mut revisions = self.revisions.lock().await; + if let Some(expected) = opts + .http_preconditions + .as_ref() + .and_then(|preconditions| preconditions.if_match.as_deref()) + { + let actual = revisions.get(&key).map(|revision| format!("memory-{revision}")); + if actual.as_deref() != Some(expected.trim_matches('"')) { + return Err(EcstoreError::PreconditionFailed); + } + } + objects.remove(&key); + revisions.remove(&key); + Ok(ObjectInfo::default()) + } + } + #[test] fn scanner_cycle_advance_fails_before_reserved_exhausted_value() { let mut cycle = CurrentCycle { @@ -5925,6 +6247,152 @@ mod tests { assert_eq!(outcome, DataUsagePersistOutcome::Failed); } + #[tokio::test] + async fn test_store_data_usage_in_backend_preserves_superseded_status() { + let store = Arc::new(MemoryConfigStore::default()); + let authoritative_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()); + let authoritative = DataUsageInfo { + scanner_epoch: Some(7), + scanner_cycle: Some(10), + ..complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 1) + }; + let authoritative_bytes = serde_json::to_vec(&authoritative).expect("authoritative snapshot should encode"); + store + .objects + .lock() + .await + .insert(authoritative_key.clone(), authoritative_bytes.clone()); + store.revisions.lock().await.insert(authoritative_key.clone(), 1); + + let (sender, receiver) = mpsc::channel(1); + sender + .send(DataUsageInfo { + last_update: Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1)), + scanner_epoch: Some(7), + scanner_cycle: Some(11), + usage_snapshot_converged: Some(false), + ..complete_usage_with_bucket_count(None, 1) + }) + .await + .expect("superseded usage snapshot should enqueue"); + drop(sender); + + let outcome = store_data_usage_in_backend_with_outcome(CancellationToken::new(), store.clone(), receiver).await; + let saved = store + .objects + .lock() + .await + .get(&memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str())) + .cloned() + .expect("superseded usage snapshot should persist"); + let saved = serde_json::from_slice::(&saved).expect("persisted usage snapshot should decode"); + + assert_eq!(outcome, DataUsagePersistOutcome::Saved); + assert!(saved.is_complete_bucket_usage_snapshot()); + assert_eq!(saved.usage_snapshot_converged, Some(false)); + assert_eq!(saved.usage_snapshot_authoritative_baseline, Some(authoritative.snapshot_identity())); + assert_eq!( + store.objects.lock().await.get(&authoritative_key), + Some(&authoritative_bytes), + "an observation must never lower the quota-authoritative snapshot" + ); + } + + #[tokio::test] + async fn test_store_data_usage_in_backend_removes_observed_after_authoritative_save() { + let store = Arc::new(MemoryConfigStore::default()); + let authoritative_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()); + let authoritative = DataUsageInfo { + scanner_epoch: Some(7), + scanner_cycle: Some(10), + ..complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 1) + }; + store.objects.lock().await.insert( + authoritative_key.clone(), + serde_json::to_vec(&authoritative).expect("authoritative snapshot should encode"), + ); + store.revisions.lock().await.insert(authoritative_key, 1); + + let (sender, receiver) = mpsc::channel(1); + sender + .send(DataUsageInfo { + last_update: Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(1)), + scanner_epoch: Some(7), + scanner_cycle: Some(11), + usage_snapshot_converged: Some(false), + ..complete_usage_with_bucket_count(None, 1) + }) + .await + .expect("superseded usage snapshot should enqueue"); + drop(sender); + + assert_eq!( + store_data_usage_in_backend_with_outcome(CancellationToken::new(), store.clone(), receiver).await, + DataUsagePersistOutcome::Saved + ); + let observed_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()); + assert!(store.objects.lock().await.contains_key(&observed_key)); + + let (sender, receiver) = mpsc::channel(1); + sender + .send(DataUsageInfo { + last_update: Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(2)), + scanner_epoch: Some(7), + scanner_cycle: Some(12), + usage_snapshot_converged: Some(true), + ..complete_usage_with_bucket_count(None, 1) + }) + .await + .expect("authoritative usage snapshot should enqueue"); + drop(sender); + + assert_eq!( + store_data_usage_in_backend_with_outcome(CancellationToken::new(), store.clone(), receiver).await, + DataUsagePersistOutcome::Saved + ); + assert!( + !store.objects.lock().await.contains_key(&observed_key), + "an authoritative snapshot should retire stale observations" + ); + + let next_authoritative = DataUsageInfo { + last_update: Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(3)), + scanner_epoch: Some(7), + scanner_cycle: Some(13), + usage_snapshot_converged: Some(true), + ..complete_usage_with_bucket_count(None, 1) + }; + let newer_observed = DataUsageInfo { + last_update: Some(std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(4)), + scanner_epoch: Some(7), + scanner_cycle: Some(14), + usage_snapshot_converged: Some(false), + usage_snapshot_authoritative_baseline: Some(next_authoritative.snapshot_identity()), + ..complete_usage_with_bucket_count(None, 1) + }; + store.objects.lock().await.insert( + observed_key.clone(), + serde_json::to_vec(&newer_observed).expect("newer observed snapshot should encode"), + ); + store.revisions.lock().await.insert(observed_key.clone(), 3); + + let (sender, receiver) = mpsc::channel(1); + sender + .send(next_authoritative) + .await + .expect("next authoritative usage snapshot should enqueue"); + drop(sender); + + assert_eq!( + store_data_usage_in_backend_with_outcome(CancellationToken::new(), store.clone(), receiver).await, + DataUsagePersistOutcome::Saved + ); + assert!( + store.objects.lock().await.contains_key(&observed_key), + "a newer observation must survive stale authoritative cleanup" + ); + } + fn mark_usage_snapshot_complete(info: &mut DataUsageInfo) { info.usage_snapshot_complete = true; } @@ -6216,13 +6684,13 @@ mod tests { #[test] #[serial] - fn finalizing_a_superseded_cycle_keeps_dirty_work_pending() { + fn finalizing_a_durable_superseded_snapshot_keeps_dirty_work_pending() { crate::scanner_io::clear_dirty_usage_bucket("photos"); crate::scanner_io::record_dirty_usage_bucket("photos"); let dirty_snapshot = crate::scanner_io::dirty_usage_buckets_for_tests(); let superseded = crate::scanner_io::ScannerCycleResult::new(ScannerCycleStatus::Superseded, Some(dirty_snapshot)); - let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(superseded, DataUsagePersistOutcome::NoUpdate); + let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(superseded, DataUsagePersistOutcome::Saved); assert_eq!(outcome, ScannerCycleOutcome::Superseded); assert!(acknowledgements.is_empty()); @@ -6452,6 +6920,49 @@ mod tests { } } + #[test] + fn superseded_retry_backoff_grows_caps_and_resets_after_convergence() { + let mut backoff = ScannerSupersededBackoff::default(); + assert_eq!(backoff.retry_interval(Duration::from_secs(24 * 60 * 60)), None); + + for expected in [60, 120, 240, 480, 960, 1_920, 3_840] { + backoff.record_cycle(ScannerCycleOutcome::Superseded); + assert_eq!( + backoff.retry_interval(Duration::from_secs(24 * 60 * 60)), + Some(Duration::from_secs(expected)) + ); + } + for _ in 0..20 { + backoff.record_cycle(ScannerCycleOutcome::Superseded); + } + assert_eq!( + backoff.retry_interval(Duration::from_secs(24 * 60 * 60)), + Some(Duration::from_secs(24 * 60 * 60)) + ); + + backoff.record_cycle(ScannerCycleOutcome::Completed); + assert_eq!(backoff.retry_interval(Duration::from_secs(24 * 60 * 60)), None); + } + + #[test] + fn superseded_retry_backoff_respects_a_faster_configured_cycle() { + let mut backoff = ScannerSupersededBackoff::default(); + backoff.record_cycle(ScannerCycleOutcome::Superseded); + + assert_eq!(backoff.retry_interval(Duration::from_secs(15)), Some(Duration::from_secs(15))); + backoff.record_cycle(ScannerCycleOutcome::Superseded); + assert_eq!(backoff.retry_interval(Duration::from_secs(15)), Some(Duration::from_secs(30))); + } + + #[test] + fn superseded_retry_backoff_grows_from_the_default_cycle() { + let mut backoff = ScannerSupersededBackoff::default(); + for expected in [60, 120, 240, 480] { + backoff.record_cycle(ScannerCycleOutcome::Superseded); + assert_eq!(backoff.retry_interval(Duration::from_secs(60)), Some(Duration::from_secs(expected))); + } + } + #[test] fn scanner_cycle_wait_plan_drives_growth_resets_and_bitrot_cap() { let runtime_config = ScannerRuntimeConfig { @@ -6529,19 +7040,23 @@ mod tests { #[test] #[serial] fn scanner_cycle_schedule_status_reports_effective_backoff() { - record_scanner_cycle_schedule(Duration::from_millis(86_400_001), true, 2_048); + record_scanner_cycle_schedule(Duration::from_millis(86_400_001), true, 2_048, true, 7); let status = scanner_cycle_schedule_status(); assert_eq!(status.effective_interval_seconds, 86_401); assert!(status.clean_idle_backoff_enabled); assert_eq!(status.clean_idle_backoff_multiplier, 2_048); + assert!(status.superseded_retry_backoff_enabled); + assert_eq!(status.superseded_cycles, 7); reset_scanner_cycle_schedule(); let status = scanner_cycle_schedule_status(); assert_eq!(status.effective_interval_seconds, 0); assert!(!status.clean_idle_backoff_enabled); assert_eq!(status.clean_idle_backoff_multiplier, 1); + assert!(!status.superseded_retry_backoff_enabled); + assert_eq!(status.superseded_cycles, 0); } #[test] @@ -6932,7 +7447,7 @@ mod tests { let mut wait = Box::pin(wait_for_next_scanner_cycle( &ctx, Duration::from_secs(60), - dirty_generation, + Some(dirty_generation), crate::runtime_config::scanner_runtime_config_generation(), crate::scanner_io::scanner_maintenance_generation(), || false, @@ -6959,7 +7474,7 @@ mod tests { let reason = wait_for_next_scanner_cycle( &ctx, Duration::from_secs(60), - dirty_generation, + Some(dirty_generation), crate::runtime_config::scanner_runtime_config_generation(), crate::scanner_io::scanner_maintenance_generation(), || false, @@ -6980,7 +7495,7 @@ mod tests { let wait = wait_for_next_scanner_cycle( &ctx, Duration::from_secs(60), - dirty_generation, + Some(dirty_generation), crate::runtime_config::scanner_runtime_config_generation(), crate::scanner_io::scanner_maintenance_generation(), || false, @@ -6992,6 +7507,25 @@ mod tests { crate::scanner_io::clear_dirty_usage_buckets_for_tests(); } + #[tokio::test(start_paused = true)] + #[serial] + async fn test_wait_for_next_scanner_cycle_can_defer_dirty_wakes_until_timer() { + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); + let ctx = CancellationToken::new(); + let wait = wait_for_next_scanner_cycle( + &ctx, + Duration::from_secs(60), + None, + crate::runtime_config::scanner_runtime_config_generation(), + crate::scanner_io::scanner_maintenance_generation(), + || false, + ); + + crate::scanner_io::record_dirty_usage_bucket("photos"); + assert_eq!(wait.await, ScannerCycleWakeReason::Timer); + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); + } + #[tokio::test] #[serial] async fn test_wait_for_next_scanner_cycle_wakes_for_repeated_dirty_bucket() { @@ -7002,7 +7536,7 @@ mod tests { let mut wait = Box::pin(wait_for_next_scanner_cycle( &ctx, Duration::from_secs(60), - dirty_generation, + Some(dirty_generation), crate::runtime_config::scanner_runtime_config_generation(), crate::scanner_io::scanner_maintenance_generation(), || false, @@ -7027,7 +7561,7 @@ mod tests { let mut wait = Box::pin(wait_for_next_scanner_cycle( &ctx, Duration::from_secs(60), - crate::scanner_io::dirty_usage_generation(), + Some(crate::scanner_io::dirty_usage_generation()), observed_generation, crate::scanner_io::scanner_maintenance_generation(), || false, @@ -7055,7 +7589,7 @@ mod tests { let mut wait = Box::pin(wait_for_next_scanner_cycle( &ctx, Duration::from_secs(60), - crate::scanner_io::dirty_usage_generation(), + Some(crate::scanner_io::dirty_usage_generation()), crate::runtime_config::scanner_runtime_config_generation(), observed_generation, || false, @@ -7077,7 +7611,7 @@ mod tests { let reason = wait_for_next_scanner_cycle( &ctx, Duration::from_secs(60), - crate::scanner_io::dirty_usage_generation(), + Some(crate::scanner_io::dirty_usage_generation()), crate::runtime_config::scanner_runtime_config_generation(), crate::scanner_io::scanner_maintenance_generation(), || true, @@ -7304,9 +7838,10 @@ mod tests { Some(Duration::from_secs(60)), &mut seen, ScannerCycleObservedGenerations { - dirty_usage: crate::scanner_io::dirty_usage_generation(), + dirty_usage: Some(crate::scanner_io::dirty_usage_generation()), runtime_config: crate::runtime_config::scanner_runtime_config_generation(), maintenance: crate::scanner_io::scanner_maintenance_generation(), + defer_cluster_activity: false, }, || false, || std::future::ready(Ok(changed.clone())), @@ -7317,6 +7852,34 @@ mod tests { assert_eq!(seen, Some(changed)); } + #[tokio::test(start_paused = true)] + #[serial] + async fn superseded_retry_wait_defers_dirty_cluster_activity_until_timer() { + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); + let ctx = CancellationToken::new(); + let mut seen = Some(BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))])); + let changed = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 8, 3))]); + + let reason = wait_for_next_scanner_cycle_with_activity( + &ctx, + Duration::from_secs(120), + Some(Duration::from_secs(60)), + &mut seen, + ScannerCycleObservedGenerations { + dirty_usage: None, + runtime_config: crate::runtime_config::scanner_runtime_config_generation(), + maintenance: crate::scanner_io::scanner_maintenance_generation(), + defer_cluster_activity: true, + }, + || false, + || std::future::ready(Ok(changed.clone())), + ) + .await; + + assert_eq!(reason, ScannerCycleWakeReason::Timer); + assert_eq!(seen, Some(changed)); + } + #[tokio::test(start_paused = true)] #[serial] async fn distributed_clean_idle_wait_blocks_backoff_for_unpropagated_maintenance() { @@ -7331,9 +7894,10 @@ mod tests { Some(Duration::from_secs(60)), &mut seen, ScannerCycleObservedGenerations { - dirty_usage: crate::scanner_io::dirty_usage_generation(), + dirty_usage: Some(crate::scanner_io::dirty_usage_generation()), runtime_config: crate::runtime_config::scanner_runtime_config_generation(), maintenance: crate::scanner_io::scanner_maintenance_generation(), + defer_cluster_activity: false, }, || false, || std::future::ready(Ok(changed.clone())), @@ -7356,9 +7920,10 @@ mod tests { Some(Duration::from_secs(60)), &mut seen, ScannerCycleObservedGenerations { - dirty_usage: crate::scanner_io::dirty_usage_generation(), + dirty_usage: Some(crate::scanner_io::dirty_usage_generation()), runtime_config: crate::runtime_config::scanner_runtime_config_generation(), maintenance: crate::scanner_io::scanner_maintenance_generation(), + defer_cluster_activity: false, }, || false, || std::future::ready(Err("node-2 is unreachable".to_string())), @@ -7383,9 +7948,10 @@ mod tests { Some(Duration::from_secs(60)), &mut seen, ScannerCycleObservedGenerations { - dirty_usage: crate::scanner_io::dirty_usage_generation(), + dirty_usage: Some(crate::scanner_io::dirty_usage_generation()), runtime_config: crate::runtime_config::scanner_runtime_config_generation(), maintenance: crate::scanner_io::scanner_maintenance_generation(), + defer_cluster_activity: false, }, || false, || std::future::ready(Ok(expected.clone())), @@ -7414,9 +7980,10 @@ mod tests { Some(Duration::from_secs(60)), &mut seen, ScannerCycleObservedGenerations { - dirty_usage: crate::scanner_io::dirty_usage_generation(), + dirty_usage: Some(crate::scanner_io::dirty_usage_generation()), runtime_config: crate::runtime_config::scanner_runtime_config_generation(), maintenance: crate::scanner_io::scanner_maintenance_generation(), + defer_cluster_activity: false, }, || false, std::future::pending::>, @@ -7445,9 +8012,10 @@ mod tests { Some(Duration::from_secs(60)), &mut seen, ScannerCycleObservedGenerations { - dirty_usage: crate::scanner_io::dirty_usage_generation(), + dirty_usage: Some(crate::scanner_io::dirty_usage_generation()), runtime_config: crate::runtime_config::scanner_runtime_config_generation(), maintenance: crate::scanner_io::scanner_maintenance_generation(), + defer_cluster_activity: false, }, || lock_lost.load(std::sync::atomic::Ordering::Acquire), std::future::pending::>, diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 2457a7cb7..efb3b5e1c 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -776,6 +776,34 @@ fn classify_nsscanner_cycle( } } +fn should_publish_usage_snapshot(status: ScannerCycleStatus) -> bool { + matches!(status, ScannerCycleStatus::Complete | ScannerCycleStatus::Superseded) +} + +fn prepare_usage_snapshot_for_publication( + status: ScannerCycleStatus, + mut data_usage_info: DataUsageInfo, +) -> Option { + if !should_publish_usage_snapshot(status) { + return None; + } + + data_usage_info.usage_snapshot_converged = Some(status == ScannerCycleStatus::Complete); + Some(data_usage_info) +} + +async fn publish_usage_snapshot( + updates: &mpsc::Sender, + status: ScannerCycleStatus, + data_usage_info: DataUsageInfo, +) -> Result { + let Some(data_usage_info) = prepare_usage_snapshot_for_publication(status, data_usage_info) else { + return Ok(false); + }; + send_data_usage_update(updates, data_usage_info).await?; + Ok(true) +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum ScannerCycleActivityStatus { Unchanged, @@ -2347,22 +2375,29 @@ impl ScannerIOCycle for ECStore { dirty_usage_status, activity_status, ); - if status != ScannerCycleStatus::Complete { + if !publish_usage_snapshot( + &updates, + status, + DataUsageInfo { + last_update: Some(SystemTime::now()), + scanner_cycle: Some(want_cycle), + usage_snapshot_complete: true, + ..Default::default() + }, + ) + .await? + { return Ok(ScannerCycleResult::new(status, None)); } - let empty_usage = DataUsageInfo { - last_update: Some(SystemTime::now()), - scanner_cycle: Some(want_cycle), - usage_snapshot_complete: true, - ..Default::default() + let dirty_usage_clear = + (status == ScannerCycleStatus::Complete).then(|| dirty_usage_snapshot.buckets.as_ref().clone()); + let remote_dirty_usage_acknowledgements = if status == ScannerCycleStatus::Complete { + crate::scanner::scanner_dirty_usage_acknowledgements(&activity_before) + } else { + Vec::new() }; - send_data_usage_update(&updates, empty_usage).await?; - let dirty_usage_clear = Some(dirty_usage_snapshot.buckets.as_ref().clone()); - return Ok( - ScannerCycleResult::new(status, dirty_usage_clear).with_remote_dirty_usage_acknowledgements( - crate::scanner::scanner_dirty_usage_acknowledgements(&activity_before), - ), - ); + return Ok(ScannerCycleResult::new(status, dirty_usage_clear) + .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)); } let total_results = expected_sources.len(); @@ -2576,10 +2611,8 @@ impl ScannerIOCycle for ECStore { dirty_usage_status, activity_status, ); - if cycle_status == ScannerCycleStatus::Complete - && let Some((data_usage_info, _)) = completed_usage - { - send_data_usage_update(&updates, data_usage_info).await?; + if let Some((data_usage_info, _)) = completed_usage { + publish_usage_snapshot(&updates, cycle_status, data_usage_info).await?; } let dirty_usage_clear = should_clear_dirty_usage_snapshot( result.is_ok(), @@ -4597,6 +4630,44 @@ mod tests { } } + #[tokio::test] + async fn structurally_complete_superseded_cycles_publish_without_claiming_convergence() { + let (updates, mut receiver) = mpsc::channel(2); + + assert!( + publish_usage_snapshot(&updates, ScannerCycleStatus::Complete, DataUsageInfo::default()) + .await + .expect("complete snapshot publication should succeed") + ); + assert!( + publish_usage_snapshot(&updates, ScannerCycleStatus::Superseded, DataUsageInfo::default()) + .await + .expect("superseded snapshot publication should succeed") + ); + assert!( + !publish_usage_snapshot(&updates, ScannerCycleStatus::Incomplete, DataUsageInfo::default()) + .await + .expect("incomplete snapshot suppression should succeed") + ); + + assert_eq!( + receiver + .recv() + .await + .expect("complete update should be sent") + .usage_snapshot_converged, + Some(true) + ); + assert_eq!( + receiver + .recv() + .await + .expect("superseded update should be sent") + .usage_snapshot_converged, + Some(false) + ); + } + #[test] fn scanner_cycle_fails_closed_for_namespace_disappearance() { for activity_status in [ diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index f3ae6b3bf..455b17085 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -58,6 +58,7 @@ pub(crate) use rustfs_ecstore::api::config::storageclass::{ RRS as ECSTORE_STORAGECLASS_RRS, STANDARD as ECSTORE_STORAGECLASS_STANDARD, }; pub(crate) use rustfs_ecstore::api::data_usage::{ + invalidate_admin_data_usage_snapshot_cache as ecstore_invalidate_admin_data_usage_snapshot_cache, invalidate_data_usage_snapshot_cache as ecstore_invalidate_data_usage_snapshot_cache, replace_bucket_usage_memory_from_info as ecstore_replace_bucket_usage_memory_from_info, }; @@ -111,11 +112,11 @@ pub(crate) mod owner { ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config, - ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_erasure_sd, - ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info, - ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_read_config, - ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config, - scanner_replication_config_for_lifecycle_eval, + ecstore_invalidate_admin_data_usage_snapshot_cache, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, + ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, + ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, + ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, + ecstore_save_config, scanner_replication_config_for_lifecycle_eval, }; #[cfg(test)] diff --git a/rustfs/src/admin/handlers/account_info.rs b/rustfs/src/admin/handlers/account_info.rs index be25c50be..ca2bda6ea 100644 --- a/rustfs/src/admin/handlers/account_info.rs +++ b/rustfs/src/admin/handlers/account_info.rs @@ -18,7 +18,7 @@ use crate::admin::runtime_sources::{current_action_credentials, object_store_fro use crate::admin::storage_api::bucket::versioning_sys::BucketVersioningSys; use crate::admin::storage_api::contract::admin::StorageAdminApi; use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions}; -use crate::admin::storage_api::data_usage::{apply_bucket_usage_memory_overlay, load_data_usage_from_backend_cached}; +use crate::admin::storage_api::data_usage::{apply_bucket_usage_memory_overlay, load_admin_data_usage_from_backend_cached}; use crate::admin::storage_api::metadata_sys; use crate::auth::get_condition_values; use crate::server::{ADMIN_PREFIX, RemoteAddr}; @@ -264,8 +264,9 @@ impl Operation for AccountInfoHandler { // process-local absolute counters to cluster-wide bucket totals. // This path never triggers a live full-version listing // (rustfs/backlog#1306); freshness is owned by the scanner. - let mut data_usage_info = map_data_usage_result(load_data_usage_from_backend_cached(store.clone()).await)?; + let mut data_usage_info = map_data_usage_result(load_admin_data_usage_from_backend_cached(store.clone()).await)?; apply_bucket_usage_memory_overlay(&mut data_usage_info).await; + account_info.usage_snapshot_converged = data_usage_info.usage_snapshot_converged; for bucket in buckets.iter() { let (rd, wr) = is_allow(bucket.name.clone()).await; diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 243ad47f3..bf82ff20c 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -784,10 +784,10 @@ pub(crate) mod data_usage { crate::storage::storage_api::ecstore_data_usage::load_data_usage_from_backend(store).await } - pub(crate) async fn load_data_usage_from_backend_cached( + pub(crate) async fn load_admin_data_usage_from_backend_cached( store: Arc, ) -> Result { - crate::storage::storage_api::ecstore_data_usage::load_data_usage_from_backend_cached(store).await + crate::storage::storage_api::ecstore_data_usage::load_admin_data_usage_from_backend_cached(store).await } } diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index c3fe65d60..c078e84c2 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -20,7 +20,9 @@ use super::storage_api::admin_usecase::capacity::{ }; use super::storage_api::admin_usecase::contract::StorageAdminApi; use super::storage_api::admin_usecase::contract::bucket::{BucketOperations as _, BucketOptions}; -use super::storage_api::admin_usecase::data_usage::{apply_bucket_usage_memory_overlay, load_data_usage_from_backend_cached}; +use super::storage_api::admin_usecase::data_usage::{ + apply_bucket_usage_memory_overlay, load_admin_data_usage_from_backend_cached, +}; use super::storage_api::admin_usecase::{ECStore, EndpointServerPools}; use crate::app::runtime_sources::{ AppContext, current_app_context, current_endpoints_handle, current_object_store_handle_for_context, @@ -265,7 +267,7 @@ impl DefaultAdminUsecase { /// This path never triggers a live full-version listing /// (rustfs/backlog#1306); freshness is owned by the scanner. pub(crate) async fn query_data_usage_info_with_store(store: Arc) -> AdminUsecaseResult { - let mut info = Self::map_data_usage_load_result(load_data_usage_from_backend_cached(store.clone()).await)?; + let mut info = Self::map_data_usage_load_result(load_admin_data_usage_from_backend_cached(store.clone()).await)?; apply_bucket_usage_memory_overlay(&mut info).await; let buckets = store .list_bucket(&BucketOptions { diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 51d50db2b..b8c154834 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -3417,6 +3417,8 @@ static IO_QUEUE_CONGESTION_WARN_THROTTLE: IoQueueCongestionWarnThrottle = IoQueu #[derive(Clone, Default)] pub struct DefaultObjectUsecase { context: Option>, + #[cfg(test)] + get_object_timeout_policy: Option, } impl DefaultObjectUsecase { @@ -3426,12 +3428,17 @@ impl DefaultObjectUsecase { #[cfg(test)] pub fn without_context() -> Self { - Self { context: None } + Self { + context: None, + get_object_timeout_policy: None, + } } pub fn from_global() -> Self { Self { context: current_app_context(), + #[cfg(test)] + get_object_timeout_policy: None, } } @@ -3440,7 +3447,22 @@ impl DefaultObjectUsecase { /// so the use-case resolves that server's store; `None` falls back to the /// ambient default. pub fn with_context(context: Option>) -> Self { - Self { context } + Self { + context, + #[cfg(test)] + get_object_timeout_policy: None, + } + } + + #[cfg(test)] + fn with_context_and_get_object_timeout_policy( + context: Option>, + get_object_timeout_policy: GetObjectTimeoutPolicy, + ) -> Self { + Self { + context, + get_object_timeout_policy: Some(get_object_timeout_policy), + } } fn bucket_metadata_sys(&self) -> Option>> { @@ -3581,7 +3603,13 @@ impl DefaultObjectUsecase { blob } - fn init_get_object_bootstrap(bucket: &str, key: &str, request_id: &str) -> S3Result { + fn init_get_object_bootstrap(&self, bucket: &str, key: &str, request_id: &str) -> S3Result { + #[cfg(test)] + let timeout_config = self + .get_object_timeout_policy + .clone() + .unwrap_or_else(GetObjectTimeoutPolicy::cached_from_env); + #[cfg(not(test))] let timeout_config = GetObjectTimeoutPolicy::cached_from_env(); let wrapper = RequestTimeoutWrapper::with_request_id(timeout_config.clone(), request_id.to_string()); let request_start = std::time::Instant::now(); @@ -5763,7 +5791,7 @@ impl DefaultObjectUsecase { context.start_time.elapsed().as_secs_f64(), ); } - let bootstrap = Self::init_get_object_bootstrap(&req.input.bucket, &req.input.key, &request_id)?; + let bootstrap = self.init_get_object_bootstrap(&req.input.bucket, &req.input.key, &request_id)?; let timeout_config = bootstrap.timeout_config; let wrapper = bootstrap.wrapper; let request_start = bootstrap.request_start; @@ -10728,7 +10756,17 @@ mod tests { .key(object.to_string()) .build() .expect("real cold-fill GET input must build"); - let usecase = DefaultObjectUsecase::with_context(Some(context)); + // The request is intentionally held behind the first producer while a + // 1.3 MiB replacement write changes its generation. Disable dynamic + // sizing for this test so runner I/O load cannot consume the five-second + // production minimum before the behavior under test is released. + let usecase = DefaultObjectUsecase::with_context_and_get_object_timeout_policy( + Some(context), + GetObjectTimeoutPolicy { + enable_dynamic_timeout: false, + ..GetObjectTimeoutPolicy::default() + }, + ); let request = tokio::spawn(async move { usecase.execute_get_object(build_request(input, Method::GET)).await }); tokio::time::timeout(Duration::from_secs(2), async { while coordinator.global_waiter_count_for_test() != 1 { diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 902520998..6c6ac7b9c 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -51,6 +51,13 @@ pub(crate) mod data_usage { crate::storage::storage_api::ecstore_data_usage::apply_bucket_usage_memory_overlay(data_usage_info).await; } + pub(crate) async fn load_admin_data_usage_from_backend_cached( + store: Arc, + ) -> Result { + crate::storage::storage_api::ecstore_data_usage::load_admin_data_usage_from_backend_cached(store).await + } + + #[cfg(test)] pub(crate) async fn load_data_usage_from_backend_cached( store: Arc, ) -> Result { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 5882e2a2b..533635fb4 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -415,15 +415,15 @@ pub(crate) mod ecstore_config { pub(crate) mod ecstore_data_usage { pub(crate) use rustfs_ecstore::api::data_usage::{ - apply_bucket_usage_memory_overlay, init_compression_total_memory_from_backend, load_data_usage_from_backend, - load_data_usage_from_backend_cached, record_bucket_delete_marker_memory, record_bucket_object_delete_memory, + apply_bucket_usage_memory_overlay, init_compression_total_memory_from_backend, load_admin_data_usage_from_backend_cached, + load_data_usage_from_backend, record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory, record_bucket_object_write_memory, record_bucket_object_write_unknown_previous_memory, store_compression_total_in_backend, }; // Test-only observables for the rustfs/backlog#1306 revert detector. #[cfg(test)] pub(crate) use rustfs_ecstore::api::data_usage::{ - compute_bucket_usage, live_bucket_usage_computations, store_data_usage_in_backend, + compute_bucket_usage, live_bucket_usage_computations, load_data_usage_from_backend_cached, store_data_usage_in_backend, }; }