diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 73400675b..17174246f 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -15,7 +15,7 @@ use super::heal_info::{classify_background_heal_read_error, decode_background_heal_info}; use super::*; use crate::EcstoreResult; -use crate::storage_api::ecstore_hold_namespace_commit; +use crate::storage_api::owner::ecstore_hold_namespace_commit; use crate::storage_api::scan::{BucketOperations as _, ObjectIO as _}; use crate::{ DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_CACHE_KEY_FORMAT, DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, @@ -1221,13 +1221,34 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin "the pending candidate must not replace the authoritative baseline" ); + let committed_body = b"committed-after-walk"; + let mut reader = PutObjReader::from_vec(committed_body.to_vec()); + store.pools[0].disk_set[0] + .put_object( + &bucket, + "object", + &mut reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("the pending tail must change the physical object before it drains"); + assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty_before); drop(pending); + let retry_budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, ScannerCycleBudgetConfig::default()); let outcome = tokio::time::timeout( Duration::from_secs(30), - run_data_scanner_cycle(&ctx, &store, &mut cycle_info, &mut revision, 1), + run_data_scanner_cycle_with_budget(&ctx, &store, &mut cycle_info, &mut revision, 1, Arc::clone(&retry_budget)), ) .await .expect("the same cycle must converge after the pending PUT drains"); + assert_eq!( + retry_budget.progress().0, + 1, + "the same-cycle retry must not reuse the pre-tail bucket cache" + ); assert!(matches!( outcome, ScannerCycleOutcome::Completed | ScannerCycleOutcome::CompletedWithPendingMaintenance @@ -1241,13 +1262,16 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin assert_eq!(usage.usage_snapshot_converged, Some(true)); assert_eq!(usage.scanner_cycle, Some(1)); assert_eq!(usage.objects_total_count, 1); - assert_eq!(usage.objects_total_size, 5); + assert_eq!( + usage.objects_total_size, + u64::try_from(committed_body.len()).expect("fixture body length") + ); let bucket_usage = usage .buckets_usage .get(&bucket) .expect("the scanned bucket should be published"); assert_eq!(bucket_usage.objects_count, 1); - assert_eq!(bucket_usage.size, 5); + assert_eq!(bucket_usage.size, u64::try_from(committed_body.len()).expect("fixture body length")); global_metrics().set_cycle(None).await; crate::scanner_io::clear_dirty_usage_buckets_for_tests(); } diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 639ca60f3..b9a9a997e 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -271,6 +271,8 @@ pub struct ScannerBucketScanPlan { all_buckets: Arc>, scope: ScannerBucketScanScope, digest: DataUsageScanPlanDigest, + // Bucket work must invalidate on namespace completion even when its scoped baseline remains reusable. + bucket_cache_digest: DataUsageScanPlanDigest, leader_epoch: u64, tier_registry_generation: u64, /// Epoch captured once for the whole scanner cycle. `None` is retained diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index a8686a1e4..e24b1471e 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -118,6 +118,7 @@ impl ScannerIOCache for SetDisks { all_buckets, scope, digest: scan_plan_digest, + bucket_cache_digest, leader_epoch, tier_registry_generation, publication_epoch, @@ -637,7 +638,7 @@ impl ScannerIOCache for SetDisks { let cache_name = path_join_buf(&[&bucket.name, DATA_USAGE_CACHE_NAME]); let bucket_scan_plan_digest = - scanner_bucket_cache_digest(scan_plan_digest, dirty_usage_buckets_clone.get(&bucket.name).copied()); + scanner_bucket_cache_digest(bucket_cache_digest, dirty_usage_buckets_clone.get(&bucket.name).copied()); if let Some(server_epoch) = remote_server_epoch { let request_sequence = remote_session_sequence; diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index bbb22885d..d1b0af3f9 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -263,6 +263,10 @@ where let activity_digest = crate::scanner::scanner_activity_snapshot_digest(&activity_before); let scan_plan_digest = scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_structural_digest(&activity_before)); + let mut bucket_cache_hasher = Sha256::new(); + bucket_cache_hasher.update(scan_plan_digest.0); + bucket_cache_hasher.update(activity_digest); + let bucket_cache_digest = DataUsageScanPlanDigest(bucket_cache_hasher.finalize().into()); let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list)); let scan_scope = resolve_scanner_bucket_scan_scope( store, @@ -412,6 +416,7 @@ where all_buckets: Arc::clone(&all_buckets), scope: scan_scope.clone(), digest: scan_plan_digest, + bucket_cache_digest, leader_epoch, tier_registry_generation, publication_epoch, diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 1b132e643..480709388 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -17,11 +17,12 @@ use super::io_disk::tier_stats_template; use super::*; use crate::scanner_budget::ScannerCycleBudgetConfig; use crate::scanner_folder::ScannerItem; +use crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot; use crate::storage_api::owner::{ EcstorePoolDecommissionInfo, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats, + ecstore_hold_namespace_commit, }; use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _}; -use crate::storage_api::{EcstoreScannerPeerDirtyUsageSnapshot, ecstore_hold_namespace_commit}; use crate::{ DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerObjectOptions, ScannerPutObjReader, UNKNOWN_TIER, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, @@ -411,12 +412,21 @@ async fn pending_put_commit_keeps_scanner_walk_live_without_authoritative_usage( } let mut pending = Some(ecstore_hold_namespace_commit(store.as_ref())); + let mut previous_activity_digest = None; + let mut structural_plan_digest = None; for (cycle, converged) in [(1, false), (2, true)] { if converged { drop(pending.take()); } assert_eq!(store.scanner_data_usage_publication_blocked().await, !converged); assert!(!store.scanner_data_movement_pause_status().await.paused); + let activity = crate::scanner::probe_scanner_activity(store.as_ref(), false) + .await + .expect("the fixture activity should be observable"); + let activity_digest = crate::scanner::scanner_activity_snapshot_digest(&activity); + if let Some(previous) = previous_activity_digest.replace(activity_digest) { + assert_ne!(previous, activity_digest, "draining a namespace commit must change the publication proof"); + } let ctx = CancellationToken::new(); let budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, ScannerCycleBudgetConfig::default()); let (updates, mut receiver) = mpsc::channel(1); @@ -435,6 +445,7 @@ async fn pending_put_commit_keeps_scanner_walk_live_without_authoritative_usage( .await .expect("namespace scanning must finish while a PUT commit is pending") .expect("namespace scanning must remain available during a pending PUT commit"); + assert_eq!(result.activity_digest(), Some(activity_digest)); if !converged { assert_eq!(budget.progress().0, 2, "the pending commit must not suppress actual object traversal"); } @@ -454,6 +465,13 @@ async fn pending_put_commit_keeps_scanner_walk_live_without_authoritative_usage( assert_eq!(usage.scanner_cycle, Some(cycle)); assert_eq!(usage.objects_total_count, 2); assert_eq!(usage.objects_total_size, 11); + assert_eq!(usage.usage_snapshot_set_states.len(), 2); + for state in &usage.usage_snapshot_set_states { + let digest = state + .scan_plan_digest + .expect("each set must retain its structural cache identity"); + assert_eq!(*structural_plan_digest.get_or_insert(digest), digest); + } let bucket_usage = usage.buckets_usage.get(&bucket).expect("the walked bucket must be present"); assert_eq!(bucket_usage.objects_count, 2); assert_eq!(bucket_usage.size, 11); diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index 4da75c6a8..a1981df30 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -114,8 +114,6 @@ pub(crate) use rustfs_ecstore::api::runtime::{ }; pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks; #[cfg(test)] -pub(crate) use rustfs_ecstore::api::set_disk::test_util::hold_namespace_commit as ecstore_hold_namespace_commit; -#[cfg(test)] pub(crate) use rustfs_ecstore::api::storage::SCANNER_PUBLICATION_LEASE_TTL_MS as ECSTORE_SCANNER_PUBLICATION_LEASE_TTL_MS; #[cfg(test)] pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx; @@ -129,6 +127,9 @@ pub(crate) use rustfs_lifecycle::{ use rustfs_storage_api as storage_contracts; pub(crate) mod owner { + #[cfg(test)] + pub(crate) use rustfs_ecstore::api::set_disk::test_util::hold_namespace_commit as ecstore_hold_namespace_commit; + pub(crate) use super::storage_contracts::{ HTTPPreconditions, HTTPRangeSpec, NS_SCANNER_PROTOCOL_VERSION, ObjectIO, ObjectOperations, ObjectToDelete, };