diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index d83da3242..2d7a2efac 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -614,6 +614,14 @@ pub mod set_disk { pub fn hold_namespace_commit(store: &crate::store::ECStore) -> impl Send + Sync { store.ctx.begin_namespace_commit() } + + /// Whether an ordinary namespace commit (a PUT rename fanout, including + /// its post-ACK tail) is still pending on this instance. Test fixtures + /// drain tails through this probe; the scanner's publication barrier no + /// longer reports them. + pub fn namespace_commits_pending(store: &crate::store::ECStore) -> bool { + store.ctx.namespace_commits_pending() + } } } diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index ee2d14a8e..4438665d6 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -468,6 +468,10 @@ impl InstanceContext { self.namespace_commit_generation.load(Ordering::Acquire) } + /// Whether a namespace commit is still pending. The scanner publication + /// barrier no longer consults it (an ordinary PUT shades its bucket + /// instead); fixtures use it to drain rename tails deterministically. + #[cfg(any(test, feature = "test-util"))] pub(crate) fn namespace_commits_pending(&self) -> bool { self.namespace_commits.load(Ordering::Acquire) != 0 || self.namespace_commit_generation() == u64::MAX } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index ba8ef8423..3786aa60d 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -3897,7 +3897,7 @@ mod tests { #[cfg(feature = "test-util")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial_test::serial(storage_class_env)] - async fn early_ack_put_tails_block_scanner_publication_until_all_renames_finish() { + async fn early_ack_put_tails_shade_buckets_without_fencing_scanner_publication() { use crate::storage_api_contracts::namespace::NamespaceLocking as _; let temp_dir = tempfile::tempdir().expect("create scanner PUT tail store dir"); @@ -3973,31 +3973,28 @@ mod tests { assert!(!active, "ordinary PUT tails are not decommission or rebalance work"); assert!(!blocked, "ordinary PUT tails must not block the movement-only scan baseline"); assert_eq!(observed_movement_generation, movement_generation); - assert!(store.scanner_data_usage_publication_blocked().await); + assert!( + !store.scanner_data_usage_publication_blocked().await, + "ordinary PUT tails shade their bucket instead of fencing the local usage publication" + ); assert!(store.scanner_data_usage_publication_admission_guard().await.is_some()); assert!(set.scanner_data_usage_publication_admission_guard().await.is_some()); - for error in [ - store - .acquire_scanner_publication_lease( - movement_generation, - crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL, - ) - .await - .expect_err("a new remote publication lease must reject pending PUT tails"), - store - .validate_scanner_publication_lease(old_lease, movement_generation) - .await - .expect_err("an existing remote lease must not bypass pending PUT tails"), + let (tail_lease, tail_generation) = store + .acquire_scanner_publication_lease(movement_generation, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL) + .await + .expect("a remote publication lease is granted while PUT tails are pending"); + assert_eq!(tail_generation, movement_generation); + store + .validate_scanner_publication_lease(old_lease, movement_generation) + .await + .expect("an existing remote lease validates across pending PUT tails"); + drop( store .acquire_scanner_publication_lease_guard(old_lease) .await - .expect_err("target-side publication admission must reject pending PUT tails"), - ] { - assert!( - error.to_string().contains("blocked"), - "publication must fail because of active tails: {error}" - ); - } + .expect("target-side publication admission accepts pending PUT tails"), + ); + assert!(store.release_scanner_publication_lease(tail_lease).await); store.release_scanner_publication_lease(old_lease).await; for (index, barrier) in barriers.iter().enumerate() { @@ -4019,7 +4016,7 @@ mod tests { assert!(store.scanner_namespace_mutation_generation() > namespace_generation); let pending = index + 1 < barriers.len(); assert_eq!(ctx.namespace_commits_pending(), pending); - assert_eq!(store.scanner_data_usage_publication_blocked().await, pending); + assert!(!store.scanner_data_usage_publication_blocked().await); assert!(!store.scanner_data_movement_activity().await.1); assert!(store.scanner_data_usage_publication_admission_guard().await.is_some()); assert!(set.scanner_data_usage_publication_admission_guard().await.is_some()); @@ -4138,7 +4135,7 @@ mod tests { #[cfg(feature = "test-util")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial_test::serial(storage_class_env)] - async fn cancelled_early_ack_put_keeps_scanner_publication_blocked_until_tail_finishes() { + async fn cancelled_early_ack_put_keeps_namespace_commit_pending_until_tail_finishes() { let temp_dir = tempfile::tempdir().expect("create cancelled scanner PUT tail store dir"); let (ctx, store, shutdown) = without_storage_class_env(build_isolated_test_store(temp_dir.path(), "scanner-cancelled-put-tail", &[4])).await; @@ -4189,7 +4186,7 @@ mod tests { assert!(tracker.running() >= 1); assert!(ctx.namespace_commits_pending()); assert!(!store.scanner_data_movement_activity().await.1); - assert!(store.scanner_data_usage_publication_blocked().await); + assert!(!store.scanner_data_usage_publication_blocked().await); assert!(store.scanner_data_usage_publication_admission_guard().await.is_some()); assert!( store.pools[0].disk_set[0] diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index d8a2385c1..f28479a13 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -928,12 +928,27 @@ impl ECStore { /// Returns whether scanner metadata may still be hidden by a local /// data-movement state. Terminal failed/canceled decommission entries /// remain suspended until an operator clears or retries them, so they are - /// a publication barrier even after the worker has stopped. Active PUT - /// rename fanouts also defer publication, including post-ACK tails. + /// a publication barrier even after the worker has stopped. + /// + /// An ordinary PUT whose rename fanout is still pending is not a barrier. + /// Its bucket is shaded through the dirty-usage generation and certified + /// per bucket by the scanner, so a sustained write load cannot hold the + /// authoritative usage snapshot back (rustfs/rustfs#8494). The scanner + /// publication leases below are movement-only for the same reason + /// (rustfs/rustfs#8305). pub async fn scanner_data_usage_publication_blocked(&self) -> bool { let operation_gate = self.ctx.data_movement_operation_gate(); let _operation_guard = operation_gate.read_owned().await; - self.scanner_data_usage_publication_snapshot_blocked().await || self.ctx.namespace_commits_pending() + self.scanner_data_usage_publication_snapshot_blocked().await + } + + /// Whether an ordinary namespace commit (a PUT rename fanout, including a + /// tail that outlives a cancelled handler) still holds its physical owner. + /// Usage publication no longer consults it; tests use it to observe owner + /// retention directly. + #[cfg(any(test, feature = "test-util"))] + pub fn namespace_commits_pending_for_test(&self) -> bool { + self.ctx.namespace_commits_pending() } pub async fn scanner_data_movement_pause_status(&self) -> ScannerDataMovementPauseStatus { @@ -1120,9 +1135,13 @@ impl ECStore { } else { ScannerPublicationLeasePurpose::Authoritative }; - // Strict leases bind the namespace generation before their async - // admission checks. An observation only needs the movement fence, but - // still fails closed if generation tracking is exhausted. + // Both purposes fence data movement only. An ordinary namespace commit + // on this node is not a reason to withhold the lease: the bucket it + // writes is shaded for the coordinator's next cycle, and a commit + // pending at every lease attempt would otherwise keep a continuously + // written peer from ever granting one (rustfs/rustfs#8305). The + // namespace generation is still recorded on an authoritative lease + // for diagnostics, and generation exhaustion still fails closed. let observed_namespace_generation = self.scanner_namespace_mutation_generation(); let operation_gate = self.ctx.data_movement_operation_gate(); let operation_guard = operation_gate.read_owned().await; @@ -1137,17 +1156,7 @@ impl ECStore { return Err(Error::other("scanner publication lease is blocked by data movement")); } - let namespace_generation = if observational_only { - None - } else { - if self.ctx.namespace_commits_pending() { - return Err(Error::other("scanner publication lease is blocked by pending namespace commit")); - } - if self.scanner_namespace_mutation_generation() != observed_namespace_generation { - return Err(Error::other("scanner publication lease namespace changed during acquisition")); - } - Some(observed_namespace_generation) - }; + let namespace_generation = (!observational_only).then_some(observed_namespace_generation); let token = Uuid::new_v4(); let expires_at = tokio::time::Instant::now() + ttl; @@ -1211,20 +1220,12 @@ impl ECStore { if self.scanner_data_movement_snapshot_locked().await.1 { return Err(Error::other("scanner publication lease is blocked by data movement")); } - let Some((lease_generation, lease_namespace_generation, purpose)) = - self.ctx.scanner_publication_lease_generations(token).await - else { + let Some((lease_generation, _, _)) = self.ctx.scanner_publication_lease_generations(token).await else { return Err(Error::other("scanner publication lease is unknown or expired")); }; - if purpose == ScannerPublicationLeasePurpose::Authoritative && self.ctx.namespace_commits_pending() { - return Err(Error::other("scanner publication lease is blocked by pending namespace commit")); - } - let namespace_generation = self.scanner_namespace_mutation_generation(); - if lease_generation != self.ctx.data_movement_generation() - || namespace_generation == u64::MAX - || (purpose == ScannerPublicationLeasePurpose::Authoritative - && lease_namespace_generation != Some(namespace_generation)) - { + // Ordinary namespace commits neither block nor stale a lease; only the + // movement generation and generation exhaustion do. + if lease_generation != self.ctx.data_movement_generation() || self.scanner_namespace_mutation_generation() == u64::MAX { return Err(Error::other("scanner publication lease generation is stale")); } Ok(()) @@ -1276,9 +1277,7 @@ impl ECStore { if self.scanner_data_movement_snapshot_locked().await.1 { return Err(Error::other("scanner publication lease is blocked by data movement")); } - let Some((lease_generation, lease_namespace_generation, purpose)) = - self.ctx.scanner_publication_lease_generations(token).await - else { + let Some((lease_generation, _, purpose)) = self.ctx.scanner_publication_lease_generations(token).await else { return Err(Error::other("scanner publication lease is unknown or expired")); }; if expected_purpose.is_some_and(|expected| expected != purpose) { @@ -1287,15 +1286,11 @@ impl ECStore { if purpose == ScannerPublicationLeasePurpose::Observational && observed_target.is_none() && !validation_only { return Err(Error::other("observational lease is limited to observed usage publication")); } - if purpose == ScannerPublicationLeasePurpose::Authoritative && self.ctx.namespace_commits_pending() { - return Err(Error::other("scanner publication lease is blocked by pending namespace commit")); - } - let namespace_generation = self.scanner_namespace_mutation_generation(); - if lease_generation != self.ctx.data_movement_generation() - || namespace_generation == u64::MAX - || (purpose == ScannerPublicationLeasePurpose::Authoritative - && lease_namespace_generation != Some(namespace_generation)) - { + // An ordinary namespace commit pending or completed on this node + // neither blocks nor stales the target admission: the bucket it + // writes is shaded for the next cycle, and the aggregate rename below + // touches only scanner-owned metadata. + if lease_generation != self.ctx.data_movement_generation() || self.scanner_namespace_mutation_generation() == u64::MAX { return Err(Error::other("scanner publication lease generation is stale")); } Ok(operation_guard) @@ -2680,11 +2675,22 @@ mod tests { let ctx = Arc::new(InstanceContext::new()); let store = build_store_with_ctx(ctx.clone()); let namespace_commit = ctx.begin_namespace_commit(); - let strict_error = store + let (strict_token, strict_generation) = store .acquire_scanner_publication_lease(0, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL) .await - .expect_err("authoritative lease must wait for namespace commits"); - assert!(strict_error.to_string().contains("pending namespace commit")); + .expect("an authoritative lease is granted while an ordinary namespace commit is pending"); + assert_eq!(strict_generation, 0); + store + .validate_scanner_publication_lease(strict_token, 0) + .await + .expect("the pending commit does not invalidate the authoritative lease"); + drop( + store + .acquire_scanner_publication_lease_guard(strict_token) + .await + .expect("target-side admission accepts the pending commit"), + ); + assert!(store.release_scanner_publication_lease(strict_token).await); let (token, movement_generation) = store .acquire_scanner_publication_lease_with_purpose(0, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL, true) @@ -2755,7 +2761,7 @@ mod tests { } #[tokio::test] - async fn scanner_publication_lease_rejects_namespace_change_during_admission() { + async fn scanner_publication_lease_survives_namespace_change_during_admission() { let ctx = Arc::new(InstanceContext::new()); let store = build_store_with_ctx(ctx.clone()); let snapshot_blocker = store.rebalance_meta.write().await; @@ -2767,19 +2773,21 @@ mod tests { assert_eq!(ctx.namespace_commit_generation(), 2); assert!(!ctx.namespace_commits_pending()); drop(snapshot_blocker); - let error = tokio::time::timeout(Duration::from_secs(1), acquire) + let (token, generation) = tokio::time::timeout(Duration::from_secs(1), acquire) .await .expect("snapshot admission must finish") - .expect_err("a namespace change during admission must not install a lease"); - assert_eq!( - error.to_string(), - "Io error: scanner publication lease namespace changed during acquisition" + .expect("an ordinary namespace commit during admission shades its bucket; it does not reject the lease"); + assert_eq!(generation, 0); + assert!( + ctx.data_movement_operation_gate().try_write_owned().is_err(), + "the installed lease holds the movement permit" ); - assert!(ctx.data_movement_operation_gate().try_write_owned().is_ok(), "no lease was installed"); + assert!(store.release_scanner_publication_lease(token).await); + assert!(ctx.data_movement_operation_gate().try_write_owned().is_ok()); } #[tokio::test] - async fn scanner_publication_lease_rechecks_namespace_after_validate_snapshot() { + async fn scanner_publication_lease_validates_across_a_completed_namespace_commit() { let ctx = Arc::new(InstanceContext::new()); let store = build_store_with_ctx(ctx.clone()); let (token, generation) = store @@ -2800,14 +2808,13 @@ mod tests { assert_eq!(ctx.namespace_commit_generation(), 2); assert!(!ctx.namespace_commits_pending()); drop(next_writer); - let error = tokio::time::timeout(Duration::from_secs(1), validate) + tokio::time::timeout(Duration::from_secs(1), validate) .await .expect("validation must finish without nesting movement read locks") - .expect_err("the final snapshot must reject a completed namespace commit"); - assert_eq!(error.to_string(), "Io error: scanner publication lease generation is stale"); + .expect("a completed ordinary namespace commit does not stale the lease"); assert!( ctx.data_movement_operation_gate().try_write_owned().is_err(), - "stale lookup retains the lease permit" + "validation retains the lease permit" ); assert!(store.release_scanner_publication_lease(token).await); assert!(ctx.data_movement_operation_gate().try_write_owned().is_ok()); diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs index 7ccdd6a09..be4560216 100644 --- a/crates/ecstore/src/store/peer.rs +++ b/crates/ecstore/src/store/peer.rs @@ -1125,15 +1125,20 @@ mod tests { .await .expect("bounded physical entry"); drop(rename); - assert!(store.scanner_data_usage_publication_blocked().await); + assert!( + !store.scanner_data_usage_publication_blocked().await, + "a pending namespace commit is not a local publication barrier" + ); assert!(ctx.namespace_commits_pending()); assert!(!sibling.namespace_commits_pending()); - assert!( + // The pending physical rename owns the namespace commit, which is + // ordinary namespace activity and no longer fences another scanner + // admission on the same token. + drop( store - .rename_local_data(&disk_ref, ("target-bucket", "staged"), &fi, ("target-bucket", "another"), Some(token)) + .acquire_scanner_publication_lease_guard(token) .await - .is_err(), - "real pending rename blocks another scanner publication" + .expect("a pending ordinary rename does not fence another scanner admission"), ); assert!(store.release_scanner_publication_lease(token).await, "remove registered token"); let gate = ctx.data_movement_operation_gate(); @@ -1290,7 +1295,7 @@ mod tests { } #[tokio::test] - async fn completed_namespace_commit_rejects_stale_scanner_target_admission() { + async fn completed_namespace_commit_keeps_scanner_target_admission() { use futures::FutureExt; use std::time::{Duration, Instant}; use tokio::time::timeout; @@ -1501,21 +1506,21 @@ mod tests { assert_eq!(final_state, (false, namespace_completed, movement_before, operation_epoch_before)); assert_eq!(user_latest.data, user_new.data); assert_eq!(user_latest.version_id, user_new.version_id); + // An ordinary commit that completed after the lease was granted is + // namespace activity the scanner certifies per bucket; it is not a + // reason to withhold the scanner-owned metadata rename. assert!( - stale_validate.is_err() && target_rename.is_err(), - "completed namespace commit must reject both stale admissions: validate={stale_validate:?}, target={:?}, metadata_changed={metadata_changed}", + stale_validate.is_ok() && target_rename.is_ok(), + "a completed ordinary commit must keep both admissions: validate={stale_validate:?}, target={:?}, metadata_changed={metadata_changed}", target_rename.as_ref().map(|_| ()), ); - assert_eq!(metadata_after, metadata_before, "stale target must not rewrite metadata"); - assert_eq!(metadata_latest.data, metadata_old.data, "latest metadata body must remain the old one"); - assert_eq!( - metadata_latest.version_id, metadata_old.version_id, - "latest metadata version must remain the old one" - ); + assert!(metadata_changed, "the admitted target publishes the staged metadata"); + assert!(latest_is_staged, "latest metadata must be the published one"); + assert_ne!(metadata_latest.data, metadata_old.data); } #[tokio::test] - async fn renewed_namespace_lease_allows_real_metadata_publication() { + async fn scanner_lease_allows_real_metadata_publication_after_an_ordinary_write() { use futures::FutureExt; use std::time::Duration; use tokio::time::timeout; @@ -1581,12 +1586,10 @@ mod tests { .expect("latest ordinary committed object"); assert_eq!(user_latest.version_id, user_new.version_id); assert_eq!(user_latest.data, user_new.data); - assert!( - store - .validate_scanner_publication_lease(old_token, movement_generation) - .await - .is_err() - ); + store + .validate_scanner_publication_lease(old_token, movement_generation) + .await + .expect("an ordinary write after the grant does not stale the lease"); assert_eq!( ctx.scanner_publication_lease_generations(old_token).await, Some(( @@ -1594,7 +1597,7 @@ mod tests { Some(namespace_generation), crate::runtime::instance::ScannerPublicationLeasePurpose::Authoritative, )), - "rejection must not refresh or discard the old lease" + "validation must not refresh or discard the lease" ); let (fresh_token, fresh_generation) = store .acquire_scanner_publication_lease(movement_generation, ttl) diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index e4c4530e3..541d3cc58 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -908,6 +908,8 @@ impl DataUsageCache { || self.info.scan_coverage_receipt.is_some() } + /// Resume and reuse under one plan; fixtures that model a single digest. + #[cfg(test)] pub(crate) fn prepare_bucket_checkpoint( &mut self, name: &str, @@ -916,6 +918,34 @@ impl DataUsageCache { source: DataUsageCacheSource, scan_plan_digest: DataUsageScanPlanDigest, identity: DataUsageScanIdentity, + ) -> DataUsageCachePrepareOutcome { + self.prepare_bucket_checkpoint_with_coverage( + name, + next_cycle, + leader_epoch, + source, + scan_plan_digest, + scan_plan_digest, + identity, + ) + } + + /// `scan_plan_digest` is the reuse plan a complete cache is current under; + /// `checkpoint_plan_digest` is the coverage plan a sweep is started and + /// resumed under (`scanner_activity_coverage_digest`). The two differ by + /// the inputs that force a new walk without changing the namespace a + /// walk traverses, so a coordinator restart invalidates finished results + /// but not a sweep in progress. + #[allow(clippy::too_many_arguments)] + pub(crate) fn prepare_bucket_checkpoint_with_coverage( + &mut self, + name: &str, + next_cycle: u64, + leader_epoch: u64, + source: DataUsageCacheSource, + scan_plan_digest: DataUsageScanPlanDigest, + checkpoint_plan_digest: DataUsageScanPlanDigest, + identity: DataUsageScanIdentity, ) -> DataUsageCachePrepareOutcome { if self.info.next_cycle > next_cycle { return DataUsageCachePrepareOutcome::RejectedNewerCycle; @@ -1045,11 +1075,11 @@ impl DataUsageCache { if let Some(progress) = &mut self.info.scan_progress && has_position { - progress.requested_plan = scan_plan_digest; + progress.requested_plan = checkpoint_plan_digest; } else { self.info.scan_progress = Some(DataUsageScanProgress { - started_plan: scan_plan_digest, - requested_plan: scan_plan_digest, + started_plan: checkpoint_plan_digest, + requested_plan: checkpoint_plan_digest, }); self.info.scan_resume_after = None; self.info.scan_checkpoint = None; diff --git a/crates/scanner/src/remote_scanner/stream.rs b/crates/scanner/src/remote_scanner/stream.rs index 659fe58c9..df250f36b 100644 --- a/crates/scanner/src/remote_scanner/stream.rs +++ b/crates/scanner/src/remote_scanner/stream.rs @@ -745,6 +745,9 @@ async fn scan_and_persist_local_bucket( "remote namespace scanner checkpoint identity unavailable during admission: {err}" )) })?, + // One digest travels on the wire: a remote sweep resumes under + // the reuse plan it was requested with. + checkpoint_plan_digest: None, }, ); match scan_state { diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index df5a6cdfc..27cf0b372 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -3645,6 +3645,13 @@ where } } +/// The local publication fence. Only storage data movement defers a +/// publication: an active or terminal decommission, a rebalance, or an +/// exhausted movement sequence. An ordinary PUT whose rename fanout is still +/// pending is not a fence; its bucket is shaded through the dirty-usage +/// generation and certified per bucket, so a sustained write load cannot +/// hold the authoritative snapshot back (rustfs/rustfs#8494, +/// rustfs/rustfs#8305). async fn scanner_local_publication_defer_reason(storeapi: &S, observational_only: bool) -> Option where S: ScannerStorage, @@ -3657,31 +3664,19 @@ where if !blocked { return None; } - // Pending namespace commits invalidate this publication attempt, but only - // storage movement creates durable, rate-limited catch-up debt. - if storeapi.scanner_data_movement_publication_blocked().await { - debug!( - target: "rustfs::scanner", - event = EVENT_SCANNER_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_RUNTIME, - stage = "local_barrier", - blocker = "data_movement", - "Scanner usage publication deferred" - ); - Some(ScannerCycleDeferReason::DataMovement) - } else { - debug!( - target: "rustfs::scanner", - event = EVENT_SCANNER_PERSIST_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_RUNTIME, - stage = "local_barrier", - blocker = "pending_namespace_commit", - "Scanner usage publication deferred" - ); - Some(ScannerCycleDeferReason::ActivityBaselineUnavailable) - } + // Both barriers describe storage movement. The authoritative barrier also + // holds terminal decommission entries an operator has not cleared, which + // is the same durable, rate-limited catch-up debt. + debug!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + stage = "local_barrier", + blocker = "data_movement", + "Scanner usage publication deferred" + ); + Some(ScannerCycleDeferReason::DataMovement) } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -4091,9 +4086,9 @@ pub use activity::scanner_topology_digest; pub(crate) use activity::{ LOCAL_SCANNER_ACTIVITY_NODE, ScannerActivityGuardrailVeto, ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, ScannerDirtyUsageAcknowledgementKind, probe_scanner_activity, scanner_activity_allows_usage_publication, - scanner_activity_dirty_usage_state_for_host, scanner_activity_guardrail_veto, scanner_activity_publication_lease_targets, - scanner_activity_snapshot_digest, scanner_activity_structural_digest, scanner_dirty_usage_acknowledgements, - scanner_peer_dirty_usage_moved, + scanner_activity_coverage_digest, scanner_activity_dirty_usage_state_for_host, scanner_activity_guardrail_veto, + scanner_activity_publication_lease_targets, scanner_activity_snapshot_digest, scanner_activity_structural_digest, + scanner_dirty_usage_acknowledgements, scanner_peer_dirty_usage_moved, }; pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance}; pub use backlog::{ diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index e018dabc9..2375a930e 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/crates/scanner/src/scanner/activity.rs @@ -1033,15 +1033,25 @@ pub(crate) fn scanner_activity_snapshot_digest(snapshot: &ScannerActivitySnapsho hasher.update(activity.dirty_usage_generation.to_be_bytes()); hasher.update([u8::from(activity.dirty_usage_pending)]); hasher.update(activity.movement_generation.to_be_bytes()); - hasher.update([u8::from(activity.publication_blocked)]); } hasher.finalize().into() } -/// Hash the activity inputs that make an existing scanner cache unsafe to -/// reuse. Regular namespace writes and dirty-usage generations are omitted: -/// their affected buckets are tracked separately and may be refreshed from a -/// complete authoritative cache baseline. +/// The reuse identity: hash the activity inputs that make an existing +/// scanner result unsafe to reuse without walking again. Besides the +/// coverage identity below it binds each peer's `instance_id` and +/// `maintenance_generation`: a restarted node may have lost the dirty hints +/// of its last journal window and a changed lifecycle or replication rule +/// needs a maintenance pass, so both invalidate the authoritative baseline +/// and the set caches and force the next cycle to walk every bucket. With +/// the bucket list it is the plan under which set caches and the +/// scoped-scan baseline are reused (`scanner_bucket_plan_digest`). +/// +/// Regular namespace writes and dirty-usage generations are omitted: their +/// buckets are tracked separately and rescanned from a complete baseline. +/// `publication_blocked` is omitted from every identity: a movement pause +/// vetoes the cycle it moves in (`scanner_activity_guardrail_veto`) and +/// describes nothing about the heap. pub(crate) fn scanner_activity_structural_digest(snapshot: &ScannerActivitySnapshot) -> [u8; 32] { let mut hasher = Sha256::new(); hasher.update(u64::try_from(snapshot.len()).unwrap_or(u64::MAX).to_be_bytes()); @@ -1057,7 +1067,41 @@ pub(crate) fn scanner_activity_structural_digest(snapshot: &ScannerActivitySnaps hasher.update(activity.topology_digest); hasher.update([u8::from(activity.data_movement_active)]); hasher.update(activity.movement_generation.to_be_bytes()); - hasher.update([u8::from(activity.publication_blocked)]); + } + hasher.finalize().into() +} + +/// The coverage identity: hash the activity inputs under which a bucket +/// sweep's traversal stays one traversal, so a checkpoint resumed in a later +/// cycle is not mixed coverage. It binds the peer set, the protocol, the +/// topology and the data-movement state. A coordinator restart, a +/// maintenance publication or a publication pause changes none of the +/// namespace a sweep walks, so `instance_id`, `maintenance_generation` and +/// `publication_blocked` are left out; hashing them made every resumed sweep +/// restart as mixed coverage after such an event (rustfs/rustfs#8494). The +/// local node's namespace and dirty generations are left out too: the +/// bucket's own dirty generation joins the checkpoint plan in +/// `scanner_bucket_cache_digest`, which is what certification compares. A +/// remote peer's generations stay in, because its dirty buckets are still +/// acknowledged per cycle rather than per bucket, and a write there between +/// the two halves of a sweep would otherwise be acknowledged without having +/// been walked. +pub(crate) fn scanner_activity_coverage_digest(snapshot: &ScannerActivitySnapshot) -> [u8; 32] { + let mut hasher = Sha256::new(); + hasher.update(u64::try_from(snapshot.len()).unwrap_or(u64::MAX).to_be_bytes()); + for (host, activity) in snapshot { + let host_bytes = host.as_bytes(); + hasher.update(u64::try_from(host_bytes.len()).unwrap_or(u64::MAX).to_be_bytes()); + hasher.update(host_bytes); + hasher.update(activity.protocol_version.to_be_bytes()); + hasher.update(activity.topology_digest); + hasher.update([u8::from(activity.data_movement_active)]); + hasher.update(activity.movement_generation.to_be_bytes()); + if host.as_str() != LOCAL_SCANNER_ACTIVITY_NODE { + hasher.update(activity.namespace_generation.to_be_bytes()); + hasher.update(activity.dirty_usage_generation.to_be_bytes()); + hasher.update([u8::from(activity.dirty_usage_pending)]); + } } hasher.finalize().into() } @@ -1107,8 +1151,10 @@ pub(crate) const SCANNER_ACTIVITY_SHADE_FIELDS: [&str; 3] = /// Compare the guardrail fields of two activity probes and name the first /// field that moved. `maintenance_generation` is a veto because no cycle today -/// distinguishes whether it published maintenance results; it is also part of -/// `scanner_activity_structural_digest`. +/// distinguishes whether it published maintenance results. The veto fields +/// are compared one by one; which of them also change the reuse or the +/// coverage identity is decided by `scanner_activity_structural_digest` and +/// `scanner_activity_coverage_digest`, not here. pub(crate) fn scanner_activity_guardrail_veto( before: &ScannerActivitySnapshot, after: &ScannerActivitySnapshot, diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 581b53eab..7b298aa4d 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -18,7 +18,9 @@ use super::heal_info::{ }; use super::*; use crate::EcstoreResult; -use crate::storage_api::owner::{WriteCommitGuard, WriteCompletion, ecstore_hold_namespace_commit}; +use crate::storage_api::owner::{ + WriteCommitGuard, WriteCompletion, ecstore_hold_namespace_commit, ecstore_namespace_commits_pending, +}; 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, @@ -1501,13 +1503,18 @@ async fn run_data_scanner_cycle_publishes_activity_for_owner_lifetime() { #[tokio::test] #[serial] -async fn coordinator_persists_only_an_observation_during_pending_put() { +async fn coordinator_publishes_authority_during_pending_put() { for drives_per_set in [1, 4] { - assert_bootstrap_observation_retry(drives_per_set).await; + assert_bootstrap_authority_during_pending_put(drives_per_set).await; } } -async fn assert_bootstrap_observation_retry(drives_per_set: usize) { +/// A PUT whose rename fanout is still pending when the walk finishes used to +/// turn the first cycle into an observation and keep authority bootstrap +/// pending until a quiet retry (rustfs/rustfs#8494). The tail is ordinary +/// namespace activity: the cycle publishes authority, discharges bootstrap, +/// and the object the tail commits afterwards is the next cycle's work. +async fn assert_bootstrap_authority_during_pending_put(drives_per_set: usize) { crate::scanner_io::clear_dirty_usage_buckets_for_tests(); let temp_dir = tempfile::tempdir().expect("bootstrap directory should be created"); let store = setup_scanner_cycle_store_at_path_with_layout(temp_dir.path(), false, 1, 1, drives_per_set).await; @@ -1559,15 +1566,14 @@ async fn assert_bootstrap_observation_retry(drives_per_set: usize) { .expect("cold object should persist before scanning"); crate::scanner_io::clear_dirty_usage_buckets_for_tests(); crate::scanner_io::record_dirty_usage_bucket(&bucket); - let dirty_before = crate::scanner_io::dirty_usage_buckets_for_tests(); let baseline = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) .await .expect("fixture usage baseline should be readable"); let pending = ecstore_hold_namespace_commit(store.as_ref()); assert_eq!( scanner_local_publication_defer_reason(store.as_ref(), false).await, - Some(ScannerCycleDeferReason::ActivityBaselineUnavailable), - "an ordinary namespace commit must not be classified as data movement" + None, + "an ordinary namespace commit is neither data movement nor a publication fence" ); assert_eq!( scanner_local_publication_defer_reason(store.as_ref(), true).await, @@ -1604,43 +1610,46 @@ async fn assert_bootstrap_observation_retry(drives_per_set: usize) { assert!( matches!( outcome, - ScannerCycleOutcome::Superseded | ScannerCycleOutcome::Partial | ScannerCycleOutcome::CompletedWithPendingMaintenance + ScannerCycleOutcome::Completed | ScannerCycleOutcome::CompletedWithPendingMaintenance ), - "a pending PUT may store an observation without publishing authority: {outcome:?}" + "a pending PUT tail must not withhold authority from a complete walk: {outcome:?}" ); finish_scanner_pause_backlog_cycle(&mut pause_backlog, &store, pause_backlog_attempt, outcome).await; let pause_backlog_status = scanner_pause_backlog_status(store.clone()).await; assert_eq!(pause_backlog_status.phase, ScannerPauseBacklogPhase::Idle); assert!(!pause_backlog_status.pending_full_scan); assert_eq!(pause_backlog_status.catch_up_attempts, 0); - assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty_before); + assert!( + !crate::scanner_io::dirty_usage_buckets_for_tests().contains_key(&bucket), + "a certified bucket leaves the dirty map" + ); assert!( matches!(revision, DataUsageCacheRevision::Etag(_)), - "the cycle-state floor should advance after its durable observation" + "the cycle-state floor should advance after its durable publication" ); - assert_eq!( - read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) - .await - .expect("the prior authoritative usage must remain readable"), - baseline, - "the pending candidate must not replace the authoritative baseline" - ); - let observed = read_config(store.clone(), DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()) + let published = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) .await - .expect("the pending cycle should persist its non-authoritative observation"); - let observed: DataUsageInfo = serde_json::from_slice(&observed).expect("observed usage should decode"); - assert_eq!(observed.usage_snapshot_converged, Some(false)); - assert!(observed.is_complete_bucket_usage_snapshot()); - assert_eq!(observed.objects_total_count, 2); - assert!(!bootstrap.record_cycle(outcome)); - assert!(bootstrap.pending(), "an observation cannot discharge authority bootstrap"); + .expect("the authoritative usage should be persisted"); + assert_ne!(published, baseline, "the complete cycle replaces the bootstrap baseline"); + let published: DataUsageInfo = serde_json::from_slice(&published).expect("published usage should decode"); + assert_eq!(published.usage_snapshot_converged, Some(true)); + assert!(!published.usage_snapshot_observational); + assert!(published.is_complete_bucket_usage_snapshot()); + assert_eq!(published.objects_total_count, 2); + assert!(bootstrap.record_cycle(outcome)); + assert!( + !bootstrap.pending(), + "a complete cycle under a pending PUT tail discharges authority bootstrap" + ); crate::storage_api::ecstore_invalidate_data_usage_snapshot_cache().await; let displayed = crate::storage_api::scanner_io::load_admin_data_usage_from_backend_cached(store.clone()) .await - .expect("admin should load the durable complete observation"); + .expect("admin should load the durable authoritative snapshot"); assert_eq!(displayed.objects_total_count, 2); - assert_eq!(displayed.usage_snapshot_converged, Some(false)); + assert_eq!(displayed.usage_snapshot_converged, Some(true)); + // The tail commits a larger body after the walk. Its bucket is shaded + // for the next cycle exactly as the production mutation observer does. let committed_body = b"committed-after-walk"; let mut reader = PutObjReader::from_vec(committed_body.to_vec()); { @@ -1658,7 +1667,7 @@ async fn assert_bootstrap_observation_retry(drives_per_set: usize) { .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); + crate::scanner_io::record_dirty_usage_bucket(&bucket); drop(pending); let retry_budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, ScannerCycleBudgetConfig::default()); let retry_cycle = cycle_info.next; @@ -1696,7 +1705,10 @@ async fn assert_bootstrap_observation_retry(drives_per_set: usize) { outcome, ScannerCycleOutcome::Completed | ScannerCycleOutcome::CompletedWithPendingMaintenance )); - assert!(bootstrap.record_cycle(outcome)); + assert!( + !bootstrap.record_cycle(outcome), + "authority bootstrap was already discharged by the first cycle" + ); assert!(!bootstrap.pending()); assert_eq!(cycle_info.next, retry_cycle + 1); assert!(!crate::scanner_io::dirty_usage_buckets_for_tests().contains_key(&bucket)); @@ -1723,7 +1735,7 @@ async fn assert_bootstrap_observation_retry(drives_per_set: usize) { #[tokio::test] #[serial] -async fn complete_cycle_keeps_strict_publication_fence_with_observational_flag() { +async fn complete_cycle_keeps_movement_fence_with_observational_flag() { let (_temp_dir, store) = setup_scanner_cycle_store().await; let pending = ecstore_hold_namespace_commit(store.as_ref()); @@ -1735,16 +1747,31 @@ async fn complete_cycle_keeps_strict_publication_fence_with_observational_flag() assert!(!complete_uses_observation, "a converged Complete snapshot is authoritative"); assert_eq!( scanner_local_publication_defer_reason(store.as_ref(), complete_uses_observation).await, - Some(ScannerCycleDeferReason::ActivityBaselineUnavailable), - "a Complete cycle must remain fenced while an ordinary namespace commit is pending" + None, + "an ordinary namespace commit shades its bucket; it is not a publication fence" ); + drop(pending); + + // A terminal decommission entry is storage movement the operator must + // clear: it fences the authoritative and the observational route alike. + store.pool_meta.write().await.pools[0].decommission = Some(crate::storage_api::owner::EcstorePoolDecommissionInfo { + failed: true, + ..Default::default() + }); + for observational_only in [false, true] { + assert_eq!( + scanner_local_publication_defer_reason(store.as_ref(), observational_only).await, + Some(ScannerCycleDeferReason::DataMovement), + "observational_only={observational_only}" + ); + } + store.pool_meta.write().await.pools[0].decommission = None; assert!(uses_observation(ScannerCycleStatus::Superseded)); assert!(uses_observation(ScannerCycleStatus::Deferred( ScannerCycleDeferReason::ActivityBaselineUnavailable ))); assert!(!uses_observation(ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement))); - drop(pending); } #[tokio::test] @@ -10126,6 +10153,7 @@ fn scoped_scan_remote_dirty_coverage_invalidates_local_bucket_current() { require_source: true, tier_registry_generation: None, checkpoint_identity: None, + checkpoint_plan_digest: None, }, ), crate::scanner_io::DataUsageCacheScanState::Prepared { @@ -10350,7 +10378,7 @@ fn scanner_activity_structural_digest_ignores_regular_bucket_writes() { } #[test] -fn scanner_activity_structural_digest_fences_restart_and_maintenance() { +fn scanner_activity_structural_digest_fences_restart_and_maintenance_but_not_publication_pauses() { let baseline = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]); let mut restarted = baseline.clone(); restarted.get_mut("node-2").expect("node should exist").instance_id = "epoch-b".to_string(); @@ -10359,7 +10387,12 @@ fn scanner_activity_structural_digest_fences_restart_and_maintenance() { .get_mut("node-2") .expect("node should exist") .maintenance_generation = 4; + let mut paused = baseline.clone(); + paused.get_mut("node-2").expect("node should exist").publication_blocked = true; + // A restarted peer may have lost its last dirty-hint window and a changed + // maintenance rule needs a maintenance pass, so both invalidate every + // reusable result (the baseline and the set caches). assert_ne!( scanner_activity_structural_digest(&baseline), scanner_activity_structural_digest(&restarted) @@ -10368,8 +10401,145 @@ fn scanner_activity_structural_digest_fences_restart_and_maintenance() { scanner_activity_structural_digest(&baseline), scanner_activity_structural_digest(&maintained) ); + // A publication pause describes nothing about the heap: it vetoes the + // cycle it moves in and is part of no identity. + assert_eq!(scanner_activity_structural_digest(&baseline), scanner_activity_structural_digest(&paused)); + assert_eq!(scanner_activity_snapshot_digest(&baseline), scanner_activity_snapshot_digest(&paused)); + assert_eq!(scanner_activity_coverage_digest(&baseline), scanner_activity_coverage_digest(&paused)); + assert_eq!( + scanner_activity_guardrail_veto(&baseline, &paused).map(|veto| veto.field), + Some("publication_blocked") + ); } +#[test] +fn scanner_activity_coverage_digest_survives_restart_and_maintenance_but_fences_the_heap() { + let baseline = BTreeMap::from([ + (LOCAL_SCANNER_ACTIVITY_NODE.to_string(), scanner_node_activity("process-1", 5, 2)), + ("node-2".to_string(), scanner_node_activity("remote-a", 7, 3)), + ]); + type Mutation = (&'static str, Box); + + // A coordinator restart, a maintenance publication, a publication pause + // or the local node's own generations do not change the namespace a + // sweep traverses, so a checkpoint resumes under the same plan; each of + // the first three is still a guardrail veto on its own. + let same_coverage: Vec = vec![ + ("instance_id", Box::new(|activity| activity.instance_id = "restarted".to_string())), + ("maintenance_generation", Box::new(|activity| activity.maintenance_generation += 1)), + ("publication_blocked", Box::new(|activity| activity.publication_blocked = true)), + ]; + for host in [LOCAL_SCANNER_ACTIVITY_NODE, "node-2"] { + for (field, mutate) in &same_coverage { + let mut changed = baseline.clone(); + mutate(changed.get_mut(host).expect("host should be present")); + assert_eq!( + scanner_activity_coverage_digest(&baseline), + scanner_activity_coverage_digest(&changed), + "{field} on {host} must not change the coverage identity" + ); + assert_eq!( + scanner_activity_guardrail_veto(&baseline, &changed).map(|veto| veto.field), + Some(*field), + "{field} on {host} still vetoes the cycle it moved in" + ); + } + } + let local_generations: Vec = vec![ + ("namespace_generation", Box::new(|activity| activity.namespace_generation += 1)), + ("dirty_usage_generation", Box::new(|activity| activity.dirty_usage_generation += 1)), + ("dirty_usage_pending", Box::new(|activity| activity.dirty_usage_pending = true)), + ]; + for (field, mutate) in &local_generations { + let mut changed = baseline.clone(); + mutate(changed.get_mut(LOCAL_SCANNER_ACTIVITY_NODE).expect("local activity")); + assert_eq!( + scanner_activity_coverage_digest(&baseline), + scanner_activity_coverage_digest(&changed), + "local {field} is certified per bucket through the bucket's own dirty generation" + ); + // A peer's generations stay in: its dirty buckets are acknowledged + // per cycle, so a write there between two halves of a sweep must + // restart the sweep. + let mut peer_changed = baseline.clone(); + mutate(peer_changed.get_mut("node-2").expect("peer activity")); + assert_ne!( + scanner_activity_coverage_digest(&baseline), + scanner_activity_coverage_digest(&peer_changed), + "remote {field} must change the coverage identity" + ); + } + let heap: Vec = vec![ + ("protocol_version", Box::new(|activity| activity.protocol_version += 1)), + ("topology_digest", Box::new(|activity| activity.topology_digest = [4; 32])), + ("data_movement_active", Box::new(|activity| activity.data_movement_active = true)), + ("movement_generation", Box::new(|activity| activity.movement_generation += 1)), + ]; + for host in [LOCAL_SCANNER_ACTIVITY_NODE, "node-2"] { + for (field, mutate) in &heap { + let mut changed = baseline.clone(); + mutate(changed.get_mut(host).expect("host should be present")); + assert_ne!( + scanner_activity_coverage_digest(&baseline), + scanner_activity_coverage_digest(&changed), + "{field} on {host} must change the coverage identity" + ); + assert_ne!( + scanner_activity_structural_digest(&baseline), + scanner_activity_structural_digest(&changed), + "{field} on {host} must change the reuse identity" + ); + } + } + let mut peer_lost = baseline.clone(); + peer_lost.remove("node-2"); + assert_ne!(scanner_activity_coverage_digest(&baseline), scanner_activity_coverage_digest(&peer_lost)); +} + +#[test] +fn bucket_checkpoint_plan_survives_a_coordinator_restart() { + let bucket_info = |name: &str| crate::storage_api::scanner_io::BucketInfo { + name: name.to_string(), + created: None, + deleted: None, + versioning: false, + object_locking: false, + }; + let buckets = vec![bucket_info("alpha"), bucket_info("beta")]; + let before = BTreeMap::from([ + (LOCAL_SCANNER_ACTIVITY_NODE.to_string(), scanner_node_activity("process-1", 5, 2)), + ("node-2".to_string(), scanner_node_activity("remote-a", 7, 3)), + ]); + let mut restarted = before.clone(); + let local = restarted + .get_mut(LOCAL_SCANNER_ACTIVITY_NODE) + .expect("local activity should exist"); + local.instance_id = "process-2".to_string(); + local.namespace_generation = 0; + local.maintenance_generation = 0; + local.dirty_usage_generation = 0; + local.dirty_usage_pending = false; + local.publication_blocked = true; + + // The plan a bucket checkpoint was started under must be the plan the + // restarted coordinator requests, or every resumed sweep restarts as + // mixed coverage and the bucket can never certify (rustfs/rustfs#8494). + // The reuse plan does change: finished results are walked again. + let coverage_plan = + |snapshot| crate::scanner_io::scanner_bucket_plan_digest(&buckets, scanner_activity_coverage_digest(snapshot)); + let reuse_plan = + |snapshot| crate::scanner_io::scanner_bucket_plan_digest(&buckets, scanner_activity_structural_digest(snapshot)); + assert_eq!(coverage_plan(&before), coverage_plan(&restarted)); + assert_ne!(reuse_plan(&before), reuse_plan(&restarted)); + + let mut extra_bucket = buckets.clone(); + extra_bucket.push(bucket_info("gamma")); + assert_ne!( + coverage_plan(&before), + crate::scanner_io::scanner_bucket_plan_digest(&extra_bucket, scanner_activity_coverage_digest(&before)), + "the bucket list stays part of the plan" + ); +} #[test] fn scanner_dirty_usage_acknowledgements_exclude_local_and_clean_nodes() { let snapshot = BTreeMap::from([ diff --git a/crates/scanner/src/scanner/tests/scoped_ack_publication.rs b/crates/scanner/src/scanner/tests/scoped_ack_publication.rs index 0acac8514..2914c985a 100644 --- a/crates/scanner/src/scanner/tests/scoped_ack_publication.rs +++ b/crates/scanner/src/scanner/tests/scoped_ack_publication.rs @@ -14,7 +14,7 @@ const PROOF_CYCLE: u64 = 11; async fn settle_namespace_commits(store: &ECStore) { tokio::time::timeout(Duration::from_secs(30), async { - while store.scanner_data_usage_publication_blocked().await { + while ecstore_namespace_commits_pending(store) { tokio::time::sleep(Duration::from_millis(1)).await; } }) diff --git a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs index 2a9aa1b60..912e57983 100644 --- a/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs +++ b/crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs @@ -948,6 +948,7 @@ async fn hot_bucket_certifies_first_full_sweep_after_mixed_end(advance_epoch: bo require_source: true, tier_registry_generation: Some(identity.tier_registry_generation), checkpoint_identity: Some(identity), + checkpoint_plan_digest: None, }, ); cache.info.skip_healing = true; @@ -1207,6 +1208,7 @@ where require_source: true, tier_registry_generation: None, checkpoint_identity: Some(identity), + checkpoint_plan_digest: None, }, ); let prepared = retained(&cache); @@ -1346,6 +1348,7 @@ where require_source: true, tier_registry_generation: Some(identity.tier_registry_generation), checkpoint_identity: Some(identity), + checkpoint_plan_digest: None, }, ); cache.info.skip_healing = true; diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index baa285a65..094ab0785 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -438,13 +438,7 @@ fn complete_scanner_cache_snapshot_plan_digest( // Completed maintenance also covers ordinary usage. Keep its exact stored // proof for cache reuse, and reject mixtures of different set work proofs. let baseline_plan_digest = DataUsageScanPlanDigest(snapshot.usage_snapshot_set_states.first()?.scan_plan_digest?); - if ![ - proof.scan_plan_digest, - scanner_bucket_work_digest(proof.scan_plan_digest, HealScanMode::Normal, true), - scanner_bucket_work_digest(proof.scan_plan_digest, HealScanMode::Deep, true), - ] - .contains(&baseline_plan_digest) - { + if !scanner_plan_digest_is_work_variant(proof.scan_plan_digest, baseline_plan_digest) { return None; } let mut states = HashSet::with_capacity(snapshot.usage_snapshot_set_states.len()); @@ -465,6 +459,32 @@ fn complete_scanner_cache_snapshot_plan_digest( (states == *proof.expected_sources).then_some(baseline_plan_digest) } +/// Whether `digest` is `structural` under any work flag: the same bucket +/// list, topology and movement state walked as a scoped, full or Deep sweep. +/// The flag changes what a walk must visit, not the namespace it describes. +fn scanner_plan_digest_is_work_variant(structural: DataUsageScanPlanDigest, digest: DataUsageScanPlanDigest) -> bool { + [ + structural, + scanner_bucket_work_digest(structural, HealScanMode::Normal, true), + scanner_bucket_work_digest(structural, HealScanMode::Deep, true), + ] + .contains(&digest) +} + +/// Buckets the last authoritative publication left pending. A `Complete` +/// cycle names every bucket whose value it could not certify, including a +/// partial walk that published the bucket's last certified value, so the +/// next scoped cycle must walk them even when the dirty map no longer names +/// them. Over-inclusion only costs a walk; a missing baseline contributes +/// nothing and the dirty map decides alone. +fn scanner_baseline_pending_buckets(proof: &ScannerCacheBaselineProof<'_>) -> Vec { + proof + .authoritative_data + .and_then(|data| serde_json::from_slice::(data).ok()) + .map(|snapshot| snapshot.usage_snapshot_pending_buckets) + .unwrap_or_default() +} + fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option { let authoritative = serde_json::from_slice::(proof.authoritative_data?).ok()?; if let Some(validated_digest) = complete_scanner_cache_snapshot_plan_digest(&authoritative, proof, None) { @@ -564,6 +584,7 @@ fn scoped_scan_scope_from_dirty_buckets( let current_buckets = all_buckets.iter().map(|bucket| bucket.name.as_str()).collect::>(); let selected_buckets = dirty_buckets .into_iter() + .chain(scanner_baseline_pending_buckets(&baseline_proof)) .filter(|bucket| current_buckets.contains(bucket.as_str())) .collect::>(); if selected_buckets.is_empty() { @@ -821,6 +842,12 @@ pub struct ScannerBucketScanPlan { digest: DataUsageScanPlanDigest, /// Includes mutation generations even when the set planner uses a structural digest. bucket_coverage_digest: DataUsageScanPlanDigest, + /// The plan a bucket sweep is started and resumed under: the coverage + /// identity (`scanner_activity_coverage_digest`) with the bucket list. + /// It leaves out what `bucket_coverage_digest` binds for reuse, so a + /// coordinator restart or a maintenance publication does not turn a + /// resumed sweep into mixed coverage. + checkpoint_coverage_digest: DataUsageScanPlanDigest, requires_full_scan: bool, service_cohort: Option>>, // Cache work must invalidate on namespace completion even when its scoped baseline remains reusable. @@ -843,10 +870,13 @@ pub struct ScannerBucketScanPlan { struct ScannerBucketFailureState { hard: Arc>>, partial: Arc>>, + /// Partial buckets whose set scan published the bucket's last certified + /// value in place of the unfinished walk; a subset of `partial`. + partial_carried: Arc>>, namespace_not_found: Arc>>, } -fn scanner_bucket_plan_digest(buckets: &[BucketInfo], activity_digest: [u8; 32]) -> DataUsageScanPlanDigest { +pub(crate) fn scanner_bucket_plan_digest(buckets: &[BucketInfo], activity_digest: [u8; 32]) -> DataUsageScanPlanDigest { let mut buckets = buckets.iter().collect::>(); buckets.sort_unstable_by(|left, right| left.name.cmp(&right.name)); @@ -953,17 +983,31 @@ fn finalize_nsscanner_result(results: &[DataUsageCache], first_err: Option ScannerBucketScanStatus { +fn scanner_bucket_scan_status( + has_failed: bool, + has_uncarried_partial: bool, + has_namespace_not_found: bool, + has_carried_partial: bool, +) -> ScannerBucketScanStatus { if has_failed { ScannerBucketScanStatus::Failed - } else if has_partial { + } else if has_uncarried_partial { ScannerBucketScanStatus::Partial } else if has_namespace_not_found { ScannerBucketScanStatus::NamespaceNotFound + } else if has_carried_partial { + ScannerBucketScanStatus::PartialCarried } else { ScannerBucketScanStatus::Complete } @@ -978,7 +1022,7 @@ fn scanner_partial_progress_is_usable( ) -> bool { bucket_plan_complete && scan_scope_matches - && bucket_status == ScannerBucketScanStatus::Partial + && matches!(bucket_status, ScannerBucketScanStatus::Partial | ScannerBucketScanStatus::PartialCarried) && !has_set_failure && !has_missing_namespace } @@ -1017,9 +1061,16 @@ fn classify_nsscanner_cycle( if activity_status == ScannerCycleActivityStatus::Unverified { return ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable); } + // A bucket walk that stopped early keeps the cycle incomplete only when + // no certified value stands in for it (`Partial`). `PartialCarried` + // means every such bucket published its last certified value and is + // reported pending; `completed_all_sets` proves the sets agreed. if budget_elapsed || cancelled - || !matches!(bucket_scan_status, ScannerBucketScanStatus::Complete) + || !matches!( + bucket_scan_status, + ScannerBucketScanStatus::Complete | ScannerBucketScanStatus::PartialCarried + ) || dirty_usage_status == DirtyUsageSnapshotStatus::Unverified { return ScannerCycleStatus::Incomplete; @@ -1030,8 +1081,9 @@ fn classify_nsscanner_cycle( // Termination is snapshot-at-the-beginning: a structurally complete walk // that the guardrail did not veto is `Complete`. Buckets whose dirty - // generation moved during the walk stay pending for the next cycle and - // are reported in the publication; they no longer supersede the cycle. + // generation moved during the walk, and buckets whose walk stopped early + // behind a certified value, stay pending for the next cycle and are + // reported in the publication; they no longer supersede the cycle. match activity_status { ScannerCycleActivityStatus::Unchanged => ScannerCycleStatus::Complete, ScannerCycleActivityStatus::Changed | ScannerCycleActivityStatus::Unverified => ScannerCycleStatus::Superseded, @@ -1131,8 +1183,10 @@ async fn scanner_cycle_activity_status( where S: ScannerStorage, { - // Read the pending-commit barrier before sampling its completion generation. - // A tail that drains during this await must invalidate the earlier baseline. + // Read the storage movement barrier before sampling the activity + // generations. A movement that starts during this await must invalidate + // the earlier baseline. Ordinary PUT tails are not part of the barrier: + // they shade their bucket and are certified per bucket below. let publication_blocked = store.scanner_data_usage_publication_blocked().await; match crate::scanner::probe_scanner_activity(store, distributed).await { Ok(after) => { diff --git a/crates/scanner/src/scanner_io/cache.rs b/crates/scanner/src/scanner_io/cache.rs index d11c2d000..df5caf093 100644 --- a/crates/scanner/src/scanner_io/cache.rs +++ b/crates/scanner/src/scanner_io/cache.rs @@ -286,6 +286,10 @@ pub(crate) struct DataUsageCacheReuseOptions { pub(crate) require_source: bool, pub(crate) tier_registry_generation: Option, pub(crate) checkpoint_identity: Option, + /// The coverage plan a sweep is started and resumed under + /// (`scanner_activity_coverage_digest`); `None` resumes under the reuse + /// plan itself, as a remote worker does with the single digest on the wire. + pub(crate) checkpoint_plan_digest: Option, } #[cfg(test)] @@ -309,6 +313,7 @@ pub(crate) fn current_cache_root_or_prepare( require_source, tier_registry_generation: None, checkpoint_identity: None, + checkpoint_plan_digest: None, }, ) } @@ -322,6 +327,7 @@ pub(crate) fn current_cache_root_or_prepare_with_generation( scan_plan_digest: DataUsageScanPlanDigest, options: DataUsageCacheReuseOptions, ) -> DataUsageCacheScanState { + let checkpoint_plan_digest = options.checkpoint_plan_digest.unwrap_or(scan_plan_digest); if cache.info.next_cycle <= next_cycle && cache.info.leader_epoch <= leader_epoch && cache.info.scan_identity != options.checkpoint_identity @@ -350,8 +356,8 @@ pub(crate) fn current_cache_root_or_prepare_with_generation( // Replication writebacks leave namespace generations unchanged. A // completed cache must start a fresh coverage sweep, even on retry. cache.info.scan_progress = Some(crate::DataUsageScanProgress { - started_plan: scan_plan_digest, - requested_plan: scan_plan_digest, + started_plan: checkpoint_plan_digest, + requested_plan: checkpoint_plan_digest, }); cache.info.scan_plan_digest = None; } @@ -368,9 +374,15 @@ pub(crate) fn current_cache_root_or_prepare_with_generation( current => DataUsageCacheScanState::Prepared { invalid_current: current.err(), outcome: match options.checkpoint_identity.filter(crate::DataUsageScanIdentity::is_valid) { - Some(identity) => { - cache.prepare_bucket_checkpoint(name, next_cycle, leader_epoch, source, scan_plan_digest, identity) - } + Some(identity) => cache.prepare_bucket_checkpoint_with_coverage( + name, + next_cycle, + leader_epoch, + source, + scan_plan_digest, + checkpoint_plan_digest, + identity, + ), None => cache.prepare_for_scan(name, next_cycle, leader_epoch, source, scan_plan_digest, options.require_source), }, }, @@ -392,6 +404,30 @@ pub(super) fn cache_snapshot_is_current( ) } +/// A set cache that stands in for buckets whose walk stopped early is not +/// the execution a full walk of the same cycle produces. Bind the carried +/// bucket names so neither snapshot is adopted in place of the other by the +/// same-execution reuse in `persist_and_publish_cache_snapshot`. +pub(super) fn scanner_carried_execution_digest( + execution_digest: DataUsageScanPlanDigest, + carried_buckets: &[String], +) -> DataUsageScanPlanDigest { + if carried_buckets.is_empty() { + return execution_digest; + } + let mut carried = carried_buckets.to_vec(); + carried.sort_unstable(); + let mut hasher = Sha256::new(); + hasher.update(b"scanner-carried-execution-v1"); + hasher.update(execution_digest.0); + hasher.update(u64::try_from(carried.len()).unwrap_or(u64::MAX).to_be_bytes()); + for bucket in carried { + hasher.update(u64::try_from(bucket.len()).unwrap_or(u64::MAX).to_be_bytes()); + hasher.update(bucket.as_bytes()); + } + DataUsageScanPlanDigest(hasher.finalize().into()) +} + /// Replication statistics are observational and may change within one /// namespace execution. Bind them without relaxing same-execution reuse. pub(super) fn scanner_replication_execution_digest( diff --git a/crates/scanner/src/scanner_io/dirty_usage.rs b/crates/scanner/src/scanner_io/dirty_usage.rs index 5e549d6c5..ed2920142 100644 --- a/crates/scanner/src/scanner_io/dirty_usage.rs +++ b/crates/scanner/src/scanner_io/dirty_usage.rs @@ -1350,7 +1350,9 @@ pub(super) fn should_clear_dirty_usage_snapshot( /// bucket whose generation differs from the snapshot's, including buckets /// dirtied after the snapshot was taken. Only a `Complete` cycle reads this, /// and a failed bucket never reaches `Complete`, so the failed set that -/// `dirty_usage_buckets_excluding_failed` removes is empty here. +/// `dirty_usage_buckets_excluding_failed` removes is empty here. The caller +/// adds the buckets whose walk stopped early behind a certified value; they +/// are pending whether or not the dirty map names them. pub(super) fn pending_dirty_usage_buckets(snapshot: &DirtyUsageBuckets) -> BTreeSet { let dirty_buckets = dirty_usage_buckets(); dirty_buckets diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index 610920362..f8a7f863b 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -36,6 +36,68 @@ pub(super) async fn collect_completed_bucket_results( } } +/// Stand in for every bucket of this set whose walk stopped early with the +/// bucket's last certified value from the previous complete set cache. +/// +/// Usage is a sum: a certified bucket's value plus an uncertified bucket's +/// last certified value is a valid snapshot (see +/// `docs/architecture/scanner-per-bucket-convergence.md`), so a walk that +/// stops early no longer makes the whole set, and with it the whole cycle, +/// incomplete. The cycle reports the bucket pending and walks it again. A +/// bucket that never certified on this set, failed, was not found, or +/// changed incarnation is not covered and keeps the set incomplete. The +/// stand-in is the previous certified value, never this walk's running +/// total. +async fn carry_partial_bucket_values( + cache: &mut DataUsageCache, + old_cache: &DataUsageCache, + old_cache_certified: bool, + buckets: &[BucketInfo], + completed_buckets: &HashSet, + bucket_failures: &ScannerBucketFailureState, + current_bucket_incarnations: Option<&HashMap>, +) -> Vec { + if !old_cache_certified { + return Vec::new(); + } + let Some(current_bucket_incarnations) = current_bucket_incarnations else { + return Vec::new(); + }; + // Snapshot each set in turn; the workers of other sets take these locks + // one at a time and must never wait behind a finalising set. + let partial = bucket_failures.partial.lock().await.clone(); + let failed = bucket_failures.hard.lock().await.clone(); + let namespace_not_found = bucket_failures.namespace_not_found.lock().await.clone(); + let mut carried = Vec::new(); + for bucket in buckets { + let name = bucket.name.as_str(); + if completed_buckets.contains(name) + || !partial.contains(name) + || failed.contains(name) + || namespace_not_found.contains(name) + { + continue; + } + let current_incarnation = current_bucket_incarnations + .get(name) + .filter(|incarnation| !incarnation.is_nil()); + let certified_incarnation = old_cache + .info + .scan_bucket_incarnations + .get(name) + .filter(|incarnation| !incarnation.is_nil()); + if current_incarnation.is_none() || certified_incarnation != current_incarnation { + continue; + } + let Some(certified) = old_cache.checked_flatten(name) else { + continue; + }; + cache.replace(name, DATA_USAGE_ROOT, certified); + carried.push(bucket.name.clone()); + } + carried +} + #[derive(Clone, Copy)] pub(super) struct ScannerSetCacheGeneration { pub(super) want_cycle: u64, @@ -211,6 +273,7 @@ impl ScannerIOCache for SetDisks { scope, digest: scan_plan_digest, bucket_coverage_digest, + checkpoint_coverage_digest, requires_full_scan, service_cohort, execution_digest, @@ -224,8 +287,10 @@ impl ScannerIOCache for SetDisks { cold_zero_walk_reuse_observed, segment_invalidation_proof, } = scan_plan; + let structural_scan_plan_digest = scan_plan_digest; let scan_plan_digest = scanner_bucket_work_digest(scan_plan_digest, scan_mode, requires_full_scan); let bucket_work_digest = scanner_bucket_work_digest(bucket_coverage_digest, scan_mode, requires_full_scan); + let bucket_checkpoint_work_digest = scanner_bucket_work_digest(checkpoint_coverage_digest, scan_mode, requires_full_scan); let pool_label = self.pool_index.to_string(); let set_label = self.set_index.to_string(); @@ -255,6 +320,9 @@ impl ScannerIOCache for SetDisks { None } }; + // Captured before `prepare_for_scan` clears the mark: only a complete + // snapshot may stand in for a bucket whose walk stops early. + let old_cache_certified = old_cache.info.snapshot_complete && old_cache.info.last_update.is_some(); let current_bucket_incarnations = scanner_current_bucket_incarnations(self.as_ref(), &all_buckets).await; let scoped_scan = prepare_scoped_set_scan( &old_cache, @@ -547,6 +615,23 @@ impl ScannerIOCache for SetDisks { old_cache.info.scan_plan_digest, ) }); + // A complete set cache written under the same structural plan with + // another work flag (a full or Deep sweep against a scoped one) + // describes the same namespace. Let `prepare_for_scan` keep its + // entries as seeds, so a bucket whose walk stops early in this + // cycle can publish its last certified value + // (`carry_partial_bucket_values`); the cycle-level baseline proof + // accepts the same variants. The LKG identity above keeps the + // digest the cache was written under. + if old_cache_certified + && old_cache.info.scan_plan_digest != Some(scan_plan_digest) + && old_cache + .info + .scan_plan_digest + .is_some_and(|digest| scanner_plan_digest_is_work_variant(structural_scan_plan_digest, digest)) + { + old_cache.info.scan_plan_digest = Some(scan_plan_digest); + } let prepare_outcome = match old_cache.prepare_for_scan( DATA_USAGE_ROOT, want_cycle, @@ -767,8 +852,12 @@ 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(bucket_work_digest, dirty_usage_buckets_clone.get(&bucket.name).copied()); + let bucket_dirty_generation = dirty_usage_buckets_clone.get(&bucket.name).copied(); + // The reuse plan a complete cache is current under, and the + // coverage plan a sweep is started and resumed under. + let bucket_scan_plan_digest = scanner_bucket_cache_digest(bucket_work_digest, bucket_dirty_generation); + let bucket_checkpoint_plan_digest = + scanner_bucket_cache_digest(bucket_checkpoint_work_digest, bucket_dirty_generation); if let Some(server_epoch) = remote_server_epoch { let request_sequence = remote_session_sequence; @@ -1072,6 +1161,7 @@ impl ScannerIOCache for SetDisks { require_source: require_cache_source, tier_registry_generation: Some(tier_registry_generation), checkpoint_identity, + checkpoint_plan_digest: Some(bucket_checkpoint_plan_digest), }, ); let outcome = match scan_state { @@ -1393,7 +1483,14 @@ impl ScannerIOCache for SetDisks { }; let partial = match scan_outcome { - ScannerDiskScanOutcome::Complete(completed_cache) => { + ScannerDiskScanOutcome::Complete(mut completed_cache) => { + // The sweep finished under its coverage plan; the + // complete cache is current under the reuse plan, + // which a restart or a maintenance publication + // changes so the bucket is walked again next time. + if completed_cache.info.snapshot_complete { + completed_cache.info.scan_plan_digest = Some(bucket_scan_plan_digest); + } cache = completed_cache; None } @@ -1633,16 +1730,53 @@ impl ScannerIOCache for SetDisks { return Err(err); } + let carried_buckets = carry_partial_bucket_values( + &mut cache, + &old_cache, + old_cache_certified, + &buckets, + &completed_buckets, + &bucket_failures, + current_bucket_incarnations.as_ref(), + ) + .await; + if !carried_buckets.is_empty() { + debug!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_SET_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + state = "partial_buckets_carried", + pool = source.pool_index, + set = source.set_index, + cycle = want_cycle, + carried_buckets = carried_buckets.len(), + first_carried_bucket = carried_buckets.iter().min(), + "Scanner set published the last certified value for buckets whose walk stopped early" + ); + bucket_failures + .partial_carried + .lock() + .await + .extend(carried_buckets.iter().cloned()); + } let completed_count = completed_buckets.len(); + let covered_count = completed_count.saturating_add(carried_buckets.len()); let coverage_complete = - should_publish_completed_snapshot(completed_count, buckets.len(), budget.budget_elapsed(), ctx.is_cancelled()); + should_publish_completed_snapshot(covered_count, buckets.len(), budget.budget_elapsed(), ctx.is_cancelled()); if coverage_complete { - let completed_segment_invalidation_proof = - scanner_completed_set_segment_invalidation_proof(&segment_invalidation_proof, cold_zero_walk_reuse_candidate); + // A carried bucket was not walked to its end, so this set cannot + // prove the producer window that prefix-level reuse would need. + let completed_segment_invalidation_proof = if carried_buckets.is_empty() { + scanner_completed_set_segment_invalidation_proof(&segment_invalidation_proof, cold_zero_walk_reuse_candidate) + } else { + None + }; cache.info.next_cycle = want_cycle; cache.info.last_update.get_or_insert_with(SystemTime::now); cache.info.snapshot_complete = true; - let replication_execution = super::cache::scanner_replication_execution_digest(execution_digest, &cache); + let carried_execution = super::cache::scanner_carried_execution_digest(execution_digest, &carried_buckets); + let replication_execution = super::cache::scanner_replication_execution_digest(carried_execution, &cache); cache.info.scan_execution_digest = Some(replication_execution); cache.info.segment_invalidation_proof = completed_segment_invalidation_proof; cache.info.lkg_snapshot_complete = false; @@ -1697,11 +1831,12 @@ impl ScannerIOCache for SetDisks { set = source.set_index, cycle = want_cycle, completed_buckets = completed_count, + carried_buckets = carried_buckets.len(), total_buckets = buckets.len(), first_incomplete_bucket = buckets .iter() .map(|bucket| bucket.name.as_str()) - .filter(|bucket| !completed_buckets.contains(*bucket)) + .filter(|bucket| !completed_buckets.contains(*bucket) && !carried_buckets.iter().any(|carried| carried == bucket)) .min(), unprocessed_buckets, budget_elapsed = budget.budget_elapsed(), diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index 58e3150e1..db55aa91c 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -397,6 +397,8 @@ where let scan_plan_digest = scanner_bucket_work_digest(structural_scan_plan_digest, scan_mode, requires_full_scan); let activity_digest = crate::scanner::scanner_activity_snapshot_digest(&activity_before); let bucket_coverage_digest = scanner_bucket_plan_digest(&all_buckets, activity_digest); + let checkpoint_coverage_digest = + scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_coverage_digest(&activity_before)); let execution_digest = scanner_bucket_work_digest(bucket_coverage_digest, scan_mode, requires_full_scan); let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list)); let scope_resolution = resolve_scanner_bucket_scan_scope( @@ -613,6 +615,7 @@ where scope: scan_scope.clone(), digest: structural_scan_plan_digest, bucket_coverage_digest, + checkpoint_coverage_digest, requires_full_scan, service_cohort: service_cohort.clone(), execution_digest, @@ -726,12 +729,15 @@ where let result = finalize_nsscanner_result(&results, first_err); let failed_buckets = bucket_failures.hard.lock().await.clone(); let partial_buckets = bucket_failures.partial.lock().await.clone(); + let carried_partial_buckets = bucket_failures.partial_carried.lock().await.clone(); let namespace_not_found_buckets = bucket_failures.namespace_not_found.lock().await.clone(); let scan_scope_matches = scanner_results_match_scan_scope(&results, &expected_sources); + let has_uncarried_partial = partial_buckets.iter().any(|bucket| !carried_partial_buckets.contains(bucket)); let bucket_scan_status = scanner_bucket_scan_status( !failed_buckets.is_empty(), - scan_scope_matches && !partial_buckets.is_empty(), + scan_scope_matches && has_uncarried_partial, scan_scope_matches && !namespace_not_found_buckets.is_empty(), + scan_scope_matches && !partial_buckets.is_empty(), ); let cold_zero_walk_oracle = scanner_cycle_cold_zero_walk_oracle( &scan_scope, @@ -811,9 +817,14 @@ where ); // Certification is per bucket: the pending set is read after the // guardrail verdict so a bucket written during the walk is reported in - // the same publication that certifies the others. + // the same publication that certifies the others. A bucket whose walk + // stopped early is pending for the same reason: the publication carries + // its last certified value, not this walk's (a `Complete` cycle has only + // carried partial buckets, see `classify_nsscanner_cycle`). let pending_buckets = if cycle_status == ScannerCycleStatus::Complete { - pending_dirty_usage_buckets(&dirty_usage_snapshot.buckets) + let mut pending = pending_dirty_usage_buckets(&dirty_usage_snapshot.buckets); + pending.extend(partial_buckets.iter().cloned()); + pending } else { ScannerCyclePendingBuckets::new() }; @@ -833,13 +844,16 @@ where } else { false }; + // A walk that stopped early certifies nothing: the bucket keeps its + // dirty generation like a failed one and is walked again next cycle. + let uncertified_buckets = failed_buckets.union(&partial_buckets).cloned().collect::>(); let dirty_usage_clear = should_clear_dirty_usage_snapshot( result.is_ok(), structurally_complete_snapshot, budget_elapsed, activity_status == ScannerCycleActivityStatus::Unchanged, &dirty_usage_snapshot.buckets, - &failed_buckets, + &uncertified_buckets, ); result?; if cycle_status == ScannerCycleStatus::Complete { diff --git a/crates/scanner/src/scanner_io/publish_gate_tests.rs b/crates/scanner/src/scanner_io/publish_gate_tests.rs index c44a16a33..388679f31 100644 --- a/crates/scanner/src/scanner_io/publish_gate_tests.rs +++ b/crates/scanner/src/scanner_io/publish_gate_tests.rs @@ -838,6 +838,7 @@ fn current_cache_root_with_new_tier_generation_resets_old_cache() { require_source: false, tier_registry_generation: Some(2), checkpoint_identity: None, + checkpoint_plan_digest: None, }, ); diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 0e495f408..29a5652d1 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -23,7 +23,7 @@ use crate::scanner_folder::ScannerItem; use crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot; use crate::storage_api::owner::{ EcstorePoolDecommissionInfo, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats, - WriteCommitGuard, WriteCompletion, ecstore_hold_namespace_commit, + WriteCommitGuard, WriteCompletion, ecstore_hold_namespace_commit, ecstore_namespace_commits_pending, }; use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _}; use crate::{ @@ -44,6 +44,8 @@ use temp_env::with_var; use time::OffsetDateTime; use uuid::Uuid; +mod checkpoint_coverage_identity; +mod partial_bucket_carry; mod per_bucket_certification; mod scoped_entry_fallback; mod service_cohort; @@ -1041,7 +1043,7 @@ async fn checkpoint_rejects_unreadable_or_changed_global_fence() { async fn wait_for_namespace_commit_tails(store: &ECStore) { tokio::time::timeout(Duration::from_secs(30), async { - while store.scanner_data_usage_publication_blocked().await { + while ecstore_namespace_commits_pending(store) { tokio::time::sleep(Duration::from_millis(1)).await; } }) @@ -1591,7 +1593,7 @@ async fn multi_pool_scanner_cycle_publishes_combined_usage() { #[tokio::test] #[serial] -async fn pending_put_commit_keeps_scanner_walk_live_without_authoritative_usage() { +async fn pending_put_commit_keeps_scanner_walk_live_and_publishes_authority() { let (_temp_dir, store) = setup_two_pool_scanner_store().await; let bucket = format!("scanner-pending-put-{}", Uuid::new_v4().simple()); store @@ -1621,11 +1623,15 @@ 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 { + for (cycle, commit_pending) in [(1, true), (2, false)] { + if !commit_pending { drop(pending.take()); } - assert_eq!(store.scanner_data_usage_publication_blocked().await, !converged); + assert_eq!(ecstore_namespace_commits_pending(store.as_ref()), commit_pending); + assert!( + !store.scanner_data_usage_publication_blocked().await, + "an ordinary PUT commit is namespace activity, not a publication barrier" + ); assert!(!store.scanner_data_movement_pause_status().await.paused); let activity = crate::scanner::probe_scanner_activity(store.as_ref(), false) .await @@ -1653,22 +1659,20 @@ async fn pending_put_commit_keeps_scanner_walk_live_without_authoritative_usage( .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 { + if commit_pending { assert_eq!(budget.progress().0, 2, "the pending commit must not suppress actual object traversal"); } assert_eq!( result.status, - if converged { - ScannerCycleStatus::Complete - } else { - ScannerCycleStatus::Superseded - } + ScannerCycleStatus::Complete, + "a pending PUT commit shades its bucket; it no longer supersedes the walk" ); let usage = receiver .recv() .await .expect("the completed walk should produce a usage candidate"); - assert_eq!(usage.usage_snapshot_converged, Some(converged)); + assert_eq!(usage.usage_snapshot_converged, Some(true)); + assert!(!usage.usage_snapshot_observational); assert_eq!(usage.scanner_cycle, Some(cycle)); assert_eq!(usage.objects_total_count, 2); assert_eq!(usage.objects_total_size, 11); @@ -2307,6 +2311,7 @@ async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers scope: ScannerBucketScanScope::default(), digest: DataUsageScanPlanDigest([6; 32]), bucket_coverage_digest: DataUsageScanPlanDigest([6; 32]), + checkpoint_coverage_digest: DataUsageScanPlanDigest([6; 32]), requires_full_scan: false, execution_digest: empty_execution, leader_epoch: 11, @@ -2400,6 +2405,63 @@ fn complete_segment_reuse_baseline( bytes::Bytes::from(serde_json::to_vec(&baseline).expect("test baseline should encode")) } +#[test] +fn scoped_scan_includes_the_baseline_pending_buckets() { + let source = DataUsageCacheSource::new(1, 2); + let expected_sources = HashSet::from([source]); + let scan_plan_digest = DataUsageScanPlanDigest([9; 32]); + let mut baseline: DataUsageInfo = + serde_json::from_slice(&complete_usage_baseline(source, scan_plan_digest, 7, 11)).expect("test baseline should decode"); + baseline.buckets_usage.insert("stale".to_string(), Default::default()); + baseline.buckets_count = 2; + baseline.usage_snapshot_converged = Some(false); + baseline.usage_snapshot_pending_buckets = vec!["stale".to_string(), "deleted".to_string()]; + baseline.usage_snapshot_pending_bucket_count = 2; + let baseline = bytes::Bytes::from(serde_json::to_vec(&baseline).expect("test baseline should encode")); + let proof = ScannerCacheBaselineProof { + authoritative_data: Some(&baseline), + observed_candidate_data: None, + expected_sources: &expected_sources, + leader_epoch: 11, + want_cycle: 8, + scan_plan_digest, + }; + let all_buckets = vec![bucket_info("photos"), bucket_info("stale"), bucket_info("cold")]; + + // A bucket the last publication left pending (a partial walk that + // published its last certified value) is walked with the dirty buckets. + let scope = scoped_scan_scope_from_dirty_buckets( + ScannerBucketScanScope::default(), + HashSet::from(["photos".to_string()]), + None, + true, + false, + &all_buckets, + proof, + ); + assert_eq!( + scope.selected_buckets_for_tests(), + Some(&HashSet::from(["photos".to_string(), "stale".to_string()])), + "pending buckets join the dirty scope; a deleted bucket is dropped" + ); + + // With nothing dirty, the pending list alone scopes the walk. + let scope = scoped_scan_scope_from_dirty_buckets( + ScannerBucketScanScope::default(), + HashSet::new(), + None, + true, + false, + &all_buckets, + proof, + ); + assert_eq!( + scope.selected_buckets_for_tests(), + Some(&HashSet::from(["stale".to_string()])), + "a pending bucket is rescanned without a dirty wake" + ); +} + #[test] fn scoped_scan_requires_a_complete_authoritative_baseline_with_exact_set_provenance() { let source = DataUsageCacheSource::new(1, 2); @@ -3721,6 +3783,73 @@ fn scanner_cycle_status_requires_a_clean_complete_snapshot() { } } +#[test] +fn carried_partial_buckets_complete_the_cycle_as_pending() { + for dirty_usage_status in [DirtyUsageSnapshotStatus::Current, DirtyUsageSnapshotStatus::Changed] { + assert_eq!( + classify_nsscanner_cycle( + true, + false, + false, + ScannerBucketScanStatus::PartialCarried, + dirty_usage_status, + ScannerCycleActivityStatus::Unchanged, + ), + ScannerCycleStatus::Complete, + "a partial walk behind a certified value is pending, not incomplete ({dirty_usage_status:?})" + ); + } + assert_eq!( + classify_nsscanner_cycle( + true, + false, + false, + ScannerBucketScanStatus::PartialCarried, + DirtyUsageSnapshotStatus::Current, + ScannerCycleActivityStatus::Changed, + ), + ScannerCycleStatus::Superseded, + "the guardrail still vetoes a carried cycle" + ); + assert_eq!( + classify_nsscanner_cycle( + true, + false, + false, + ScannerBucketScanStatus::PartialCarried, + DirtyUsageSnapshotStatus::Current, + ScannerCycleActivityStatus::Unverified, + ), + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable) + ); + for (completed_all_sets, budget, cancelled, bucket_status) in [ + (false, false, false, ScannerBucketScanStatus::PartialCarried), + (true, true, false, ScannerBucketScanStatus::PartialCarried), + (true, false, true, ScannerBucketScanStatus::PartialCarried), + (true, false, false, ScannerBucketScanStatus::Partial), + ] { + assert_eq!( + classify_nsscanner_cycle( + completed_all_sets, + budget, + cancelled, + bucket_status, + DirtyUsageSnapshotStatus::Current, + ScannerCycleActivityStatus::Unchanged, + ), + ScannerCycleStatus::Incomplete, + "a set that could not stand in, a budget, a cancellation, or an uncovered partial walk stays incomplete \ + ({completed_all_sets}, {budget}, {cancelled}, {bucket_status:?})" + ); + } + for bucket_status in [ScannerBucketScanStatus::Partial, ScannerBucketScanStatus::PartialCarried] { + assert!( + scanner_partial_progress_is_usable(true, true, bucket_status, false, false), + "{bucket_status:?} saved checkpoints the retry can resume" + ); + } +} + #[test] fn checkpoint_fixture_superseded_is_distinct_from_partial_and_cancel() { for (budget, cancelled, bucket, expected) in [ @@ -3940,12 +4069,41 @@ fn scanner_cycle_fails_closed_when_dirty_generation_is_unverified() { ); } +#[test] +fn carried_execution_identity_differs_from_a_full_walk_and_is_order_independent() { + let execution = DataUsageScanPlanDigest([7; 32]); + assert_eq!(super::cache::scanner_carried_execution_digest(execution, &[]), execution); + let carried = super::cache::scanner_carried_execution_digest(execution, &["b".to_string(), "a".to_string()]); + assert_ne!(carried, execution, "a set that stood in for a bucket is not the full walk's execution"); + assert_eq!( + carried, + super::cache::scanner_carried_execution_digest(execution, &["a".to_string(), "b".to_string()]) + ); + assert_ne!(carried, super::cache::scanner_carried_execution_digest(execution, &["a".to_string()])); +} + #[test] fn scanner_bucket_failure_status_preserves_the_strongest_failure() { - assert_eq!(scanner_bucket_scan_status(false, false, false), ScannerBucketScanStatus::Complete); - assert_eq!(scanner_bucket_scan_status(false, false, true), ScannerBucketScanStatus::NamespaceNotFound); - assert_eq!(scanner_bucket_scan_status(false, true, true), ScannerBucketScanStatus::Partial); - assert_eq!(scanner_bucket_scan_status(true, true, true), ScannerBucketScanStatus::Failed); + assert_eq!(scanner_bucket_scan_status(false, false, false, false), ScannerBucketScanStatus::Complete); + assert_eq!( + scanner_bucket_scan_status(false, false, true, false), + ScannerBucketScanStatus::NamespaceNotFound + ); + assert_eq!(scanner_bucket_scan_status(false, true, true, false), ScannerBucketScanStatus::Partial); + assert_eq!(scanner_bucket_scan_status(true, true, true, false), ScannerBucketScanStatus::Failed); + // A partial walk covered by its last certified value may complete, but + // never ahead of an uncovered partial walk, a failure, or a missing + // namespace. + assert_eq!( + scanner_bucket_scan_status(false, false, false, true), + ScannerBucketScanStatus::PartialCarried + ); + assert_eq!(scanner_bucket_scan_status(false, true, false, true), ScannerBucketScanStatus::Partial); + assert_eq!( + scanner_bucket_scan_status(false, false, true, true), + ScannerBucketScanStatus::NamespaceNotFound + ); + assert_eq!(scanner_bucket_scan_status(true, false, false, true), ScannerBucketScanStatus::Failed); } #[test] diff --git a/crates/scanner/src/scanner_io/tests/checkpoint_coverage_identity.rs b/crates/scanner/src/scanner_io/tests/checkpoint_coverage_identity.rs new file mode 100644 index 000000000..d6490f1cf --- /dev/null +++ b/crates/scanner/src/scanner_io/tests/checkpoint_coverage_identity.rs @@ -0,0 +1,171 @@ +// Copyright 2026 RustFS Team +// SPDX-License-Identifier: Apache-2.0 + +//! A bucket sweep interrupted by the cycle budget resumes from its checkpoint +//! in a later cycle even after an event that changes the reuse identity but +//! not the namespace. The event here is a maintenance publication; a +//! coordinator restart is the same case for the coverage identity, but a +//! process cannot change its own instance id in a test. The other buckets +//! are walked again; the sweep in progress is not restarted as mixed +//! coverage (rustfs/rustfs#8494). + +use super::*; + +const HOT_OBJECTS: [&str; 6] = ["p0/a", "p1/b", "p2/c", "p3/d", "p4/e", "p5/f"]; +const LEADER_EPOCH: u64 = 11; + +async fn put_fixture_object(store: &Arc, bucket: &str, object: &str) { + let mut reader = ScannerPutObjReader::from_vec(vec![7; 16]); + store + .put_object(bucket, object, &mut reader, &ScannerObjectOptions::default()) + .await + .expect("fixture object should persist"); +} + +async fn run_full_cycle( + store: &Arc, + cycle: u64, + max_objects: Option, +) -> (ScannerCycleResult, Option, Arc) { + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new_with_progress_tracking( + &ctx, + ScannerCycleBudgetConfig { + max_objects, + ..Default::default() + }, + ); + let (updates, mut receiver) = mpsc::channel(1); + let result = tokio::time::timeout( + Duration::from_secs(60), + nsscanner_with_storage_status_scoped( + store.as_ref(), + ScannerCycleRequest { + ctx: budget.token(), + budget: Arc::clone(&budget), + updates, + want_cycle: cycle, + leader_epoch: LEADER_EPOCH, + scan_mode: HealScanMode::Normal, + scan_scope: ScannerBucketScanScope::default(), + persisted_usage_baseline: None, + observed_usage_candidate: None, + requires_full_scan: true, + service_cohort: None, + resolved_scope_observer: None, + post_walk_gate: None, + }, + ), + ) + .await + .expect("cycle should finish within the test deadline") + .expect("cycle should succeed"); + let snapshot = receiver.recv().await; + assert!(receiver.recv().await.is_none(), "a cycle publishes at most one terminal snapshot"); + (result, snapshot, budget) +} + +#[tokio::test] +#[serial] +async fn checkpoint_resumes_across_a_maintenance_publication() { + let (_temp_dir, store) = setup_local_scanner_store(1, 1).await; + clear_dirty_usage_buckets_for_tests(); + for bucket in ["hot-bucket", "idle-bucket"] { + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + } + for object in HOT_OBJECTS { + put_fixture_object(&store, "hot-bucket", object).await; + } + put_fixture_object(&store, "idle-bucket", "p0/a").await; + record_dirty_usage_bucket("hot-bucket"); + record_dirty_usage_bucket("idle-bucket"); + wait_for_namespace_commit_tails(store.as_ref()).await; + + // Cycle 1: a quiet full walk certifies both buckets. + let (result, first, _) = run_full_cycle(&store, 1, None).await; + let first = first.expect("a complete cycle publishes"); + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(first.usage_snapshot_converged, Some(true)); + assert_eq!(first.buckets_usage["hot-bucket"].objects_count, 6); + assert!(result.clear_verified_usage().is_empty()); + assert!(dirty_usage_buckets_for_tests().is_empty()); + + // Cycle 2: the written bucket is the only dirty one, so it is walked + // first and the object budget stops its sweep with a checkpoint; the + // cycle itself is incomplete. + for object in ["p6/g", "p7/h"] { + put_fixture_object(&store, "hot-bucket", object).await; + } + record_dirty_usage_bucket("hot-bucket"); + wait_for_namespace_commit_tails(store.as_ref()).await; + let (result, _, budget) = run_full_cycle(&store, 2, Some(2)).await; + assert_eq!(result.status, ScannerCycleStatus::Incomplete); + assert!(budget.budget_elapsed()); + let set = Arc::clone(&store.pools[0].disk_set[0]); + let mut checkpoint = DataUsageCache::default(); + checkpoint + .load(Arc::clone(&set), &path_join_buf(&["hot-bucket", DATA_USAGE_CACHE_NAME])) + .await + .expect("the interrupted sweep should leave a bucket cache"); + assert!(!checkpoint.info.snapshot_complete); + assert!(checkpoint.has_scan_position(), "the interrupted sweep must keep a durable position"); + let started_plan = checkpoint + .info + .scan_progress + .expect("the interrupted sweep records its coverage plan") + .started_plan; + + // A maintenance publication between the cycles changes the reuse + // identity (finished results are walked again) but not the coverage + // identity (a sweep in progress keeps its plan). + let before = crate::scanner::probe_scanner_activity(store.as_ref(), false) + .await + .expect("the fixture activity should be observable"); + record_scanner_maintenance_change("idle-bucket"); + let after = crate::scanner::probe_scanner_activity(store.as_ref(), false) + .await + .expect("the fixture activity should be observable"); + assert_ne!( + crate::scanner::scanner_activity_structural_digest(&before), + crate::scanner::scanner_activity_structural_digest(&after), + "a maintenance publication invalidates reusable results" + ); + assert_eq!( + crate::scanner::scanner_activity_coverage_digest(&before), + crate::scanner::scanner_activity_coverage_digest(&after), + "a maintenance publication keeps the coverage plan" + ); + + // Cycle 3: the hot bucket resumes from its checkpoint and certifies; + // a fresh walk would have read all nine objects. + let (result, third, budget) = run_full_cycle(&store, 3, None).await; + let third = third.expect("a complete cycle publishes"); + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(result.pending_dirty_usage_buckets(), 0); + assert_eq!(third.usage_snapshot_converged, Some(true)); + assert_eq!(third.buckets_usage["hot-bucket"].objects_count, 8); + assert_eq!(third.buckets_usage["idle-bucket"].objects_count, 1); + assert!( + budget.progress().0 < 9, + "the resumed sweep must not restart from the first object: {} objects read", + budget.progress().0 + ); + let mut completed = DataUsageCache::default(); + completed + .load(set, &path_join_buf(&["hot-bucket", DATA_USAGE_CACHE_NAME])) + .await + .expect("the certified bucket cache should load"); + assert!(completed.info.snapshot_complete); + assert!(completed.info.scan_progress.is_none()); + assert_ne!( + completed.info.scan_plan_digest, + Some(started_plan), + "a complete cache is current under the reuse plan, not the coverage plan" + ); + assert!(result.clear_verified_usage().is_empty()); + assert!(dirty_usage_buckets_for_tests().is_empty()); + clear_dirty_usage_buckets_for_tests(); +} diff --git a/crates/scanner/src/scanner_io/tests/partial_bucket_carry.rs b/crates/scanner/src/scanner_io/tests/partial_bucket_carry.rs new file mode 100644 index 000000000..bd1fec3ad --- /dev/null +++ b/crates/scanner/src/scanner_io/tests/partial_bucket_carry.rs @@ -0,0 +1,270 @@ +// Copyright 2026 RustFS Team +// SPDX-License-Identifier: Apache-2.0 + +//! A bucket whose walk stops early without exhausting the cycle budget (here +//! a sweep resumed under a newer plan, the `mixed_coverage` case of +//! rustfs/rustfs#8494) no longer makes the whole cycle `Incomplete`. The set +//! publishes the bucket's last certified value, the cycle completes with the +//! bucket pending, and the next cycle walks it again. A bucket that never +//! certified keeps the cycle incomplete. + +use super::*; + +const BUCKETS: [&str; 3] = ["cold-bucket", "hot-bucket", "idle-bucket"]; +const LEADER_EPOCH: u64 = 11; + +async fn put_fixture_object(store: &Arc, bucket: &str, object: &str) { + let mut reader = ScannerPutObjReader::from_vec(vec![7; 16]); + store + .put_object(bucket, object, &mut reader, &ScannerObjectOptions::default()) + .await + .expect("fixture object should persist"); +} + +struct CycleOptions { + baseline: Option, + requires_full_scan: bool, + max_objects: Option, + resolved_scope_observer: Option>, +} + +/// Run one cycle; an incomplete cycle may publish nothing or an observation. +async fn run_cycle(store: &Arc, cycle: u64, options: CycleOptions) -> (ScannerCycleResult, Option) { + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new_with_progress_tracking( + &ctx, + ScannerCycleBudgetConfig { + max_objects: options.max_objects, + ..Default::default() + }, + ); + let (updates, mut receiver) = mpsc::channel(1); + let result = tokio::time::timeout( + Duration::from_secs(60), + nsscanner_with_storage_status_scoped( + store.as_ref(), + ScannerCycleRequest { + ctx: budget.token(), + budget, + updates, + want_cycle: cycle, + leader_epoch: LEADER_EPOCH, + scan_mode: HealScanMode::Normal, + scan_scope: ScannerBucketScanScope::default(), + persisted_usage_baseline: options.baseline, + observed_usage_candidate: None, + requires_full_scan: options.requires_full_scan, + service_cohort: None, + resolved_scope_observer: options.resolved_scope_observer, + post_walk_gate: None, + }, + ), + ) + .await + .expect("cycle should finish within the test deadline") + .expect("cycle should succeed"); + let snapshot = receiver.recv().await; + assert!(receiver.recv().await.is_none(), "a cycle publishes at most one terminal snapshot"); + (result, snapshot) +} + +fn baseline_bytes(snapshot: &DataUsageInfo) -> Bytes { + Bytes::from(serde_json::to_vec(snapshot).expect("snapshot should encode")) +} + +fn complete_cycle(requires_full_scan: bool, baseline: Option<&DataUsageInfo>) -> CycleOptions { + CycleOptions { + baseline: baseline.map(baseline_bytes), + requires_full_scan, + max_objects: None, + resolved_scope_observer: None, + } +} + +#[tokio::test] +#[serial] +async fn partial_bucket_with_a_certified_value_completes_the_cycle_as_pending() { + let (_temp_dir, store) = setup_local_scanner_store(1, 1).await; + clear_dirty_usage_buckets_for_tests(); + for bucket in BUCKETS { + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + put_fixture_object(&store, bucket, "p0/initial").await; + record_dirty_usage_bucket(bucket); + } + wait_for_namespace_commit_tails(store.as_ref()).await; + + // Cycle 1: a quiet full walk certifies every bucket. + let (result, first) = run_cycle(&store, 1, complete_cycle(true, None)).await; + let first = first.expect("a complete cycle publishes"); + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(first.usage_snapshot_converged, Some(true)); + assert!(result.clear_verified_usage().is_empty()); + assert!(dirty_usage_buckets_for_tests().is_empty()); + + // Cycle 2: a scoped walk of the written bucket leaves the set cache + // under the scoped plan, which the full walk below must still seed from. + for object in ["p1/second", "p2/second", "p3/second"] { + put_fixture_object(&store, "hot-bucket", object).await; + } + record_dirty_usage_bucket("hot-bucket"); + wait_for_namespace_commit_tails(store.as_ref()).await; + let (result, second) = run_cycle(&store, 2, complete_cycle(false, Some(&first))).await; + let second = second.expect("a complete cycle publishes"); + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(second.usage_snapshot_converged, Some(true)); + assert_eq!(second.buckets_usage["hot-bucket"].objects_count, 4); + assert!(result.clear_verified_usage().is_empty()); + assert!(dirty_usage_buckets_for_tests().is_empty()); + + // Cycle 3: the object budget stops the hot bucket's walk and leaves a + // checkpoint under this cycle's plan; the cycle itself is incomplete. + for object in ["p4/third", "p5/third"] { + put_fixture_object(&store, "hot-bucket", object).await; + } + record_dirty_usage_bucket("hot-bucket"); + wait_for_namespace_commit_tails(store.as_ref()).await; + let (result, _) = run_cycle( + &store, + 3, + CycleOptions { + baseline: Some(baseline_bytes(&second)), + requires_full_scan: false, + max_objects: Some(2), + resolved_scope_observer: None, + }, + ) + .await; + assert_eq!(result.status, ScannerCycleStatus::Incomplete, "an exhausted budget stays incomplete"); + let set = Arc::clone(&store.pools[0].disk_set[0]); + let mut interrupted = DataUsageCache::default(); + interrupted + .load(set, &path_join_buf(&["hot-bucket", DATA_USAGE_CACHE_NAME])) + .await + .expect("the interrupted sweep should leave a bucket cache"); + assert!(!interrupted.info.snapshot_complete); + assert!(interrupted.has_scan_position(), "the interrupted sweep keeps a durable position"); + + // Cycle 4: the checkpoint resumes under a different plan, so the sweep + // is mixed coverage and returns Partial without exhausting any budget. + // The set stands in with the last certified value and the cycle + // completes with the bucket pending instead of being incomplete. + let (result, fourth) = run_cycle(&store, 4, complete_cycle(true, Some(&second))).await; + let fourth = fourth.expect("a complete cycle publishes"); + assert_eq!( + result.status, + ScannerCycleStatus::Complete, + "a partial walk behind a certified value no longer makes the cycle incomplete" + ); + assert_eq!(result.pending_dirty_usage_buckets(), 1); + assert!(!fourth.usage_snapshot_observational, "the publication stays authoritative"); + assert_eq!(fourth.usage_snapshot_converged, Some(false)); + assert_eq!(fourth.usage_snapshot_pending_buckets, vec!["hot-bucket".to_string()]); + assert_eq!(fourth.usage_snapshot_pending_bucket_count, 1); + assert!(fourth.is_complete_bucket_usage_snapshot()); + assert_eq!( + fourth.buckets_usage["hot-bucket"].objects_count, 4, + "the pending bucket carries its last certified value, not the partial walk" + ); + assert_eq!(fourth.buckets_usage["cold-bucket"].objects_count, 1); + assert_eq!(fourth.buckets_usage["idle-bucket"].objects_count, 1); + assert_eq!(fourth.objects_total_count, 6); + assert!(result.clear_verified_usage().is_empty()); + assert_eq!( + dirty_usage_buckets_for_tests().keys().cloned().collect::>(), + vec!["hot-bucket".to_string()], + "a partial walk certifies nothing: the bucket stays dirty" + ); + + // Cycle 5: even with the dirty map gone, the baseline's pending list + // scopes the walk to the bucket, whose fresh sweep certifies it. + clear_dirty_usage_buckets_for_tests(); + let (observer, resolved_scope) = tokio::sync::oneshot::channel(); + let (result, fifth) = run_cycle( + &store, + 5, + CycleOptions { + baseline: Some(baseline_bytes(&fourth)), + requires_full_scan: false, + max_objects: None, + resolved_scope_observer: Some(observer), + }, + ) + .await; + let fifth = fifth.expect("a complete cycle publishes"); + let scope = resolved_scope.await.expect("the cycle should report its resolved scope"); + assert_eq!( + scope.selected_buckets_for_tests(), + Some(&HashSet::from(["hot-bucket".to_string()])), + "the next cycle walks the pending bucket" + ); + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(result.pending_dirty_usage_buckets(), 0); + assert_eq!(fifth.usage_snapshot_converged, Some(true)); + assert!(fifth.usage_snapshot_pending_buckets.is_empty()); + assert_eq!(fifth.buckets_usage["hot-bucket"].objects_count, 6); + assert_eq!(fifth.buckets_usage["cold-bucket"].objects_count, 1); + assert_eq!(fifth.objects_total_count, 8); + assert!(result.clear_verified_usage().is_empty()); + assert!(dirty_usage_buckets_for_tests().is_empty()); + clear_dirty_usage_buckets_for_tests(); +} + +#[tokio::test] +#[serial] +async fn partial_bucket_without_a_certified_value_keeps_the_cycle_incomplete() { + let (_temp_dir, store) = setup_local_scanner_store(1, 1).await; + clear_dirty_usage_buckets_for_tests(); + store + .make_bucket("fresh-bucket", &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + for object in ["p0/a", "p1/b", "p2/c", "p3/d"] { + put_fixture_object(&store, "fresh-bucket", object).await; + } + record_dirty_usage_bucket("fresh-bucket"); + wait_for_namespace_commit_tails(store.as_ref()).await; + + // Cycle 1: the budget stops the only bucket's first walk; no set cache + // was ever published, so nothing can stand in for the bucket. + let (result, _) = run_cycle( + &store, + 1, + CycleOptions { + baseline: None, + requires_full_scan: true, + max_objects: Some(2), + resolved_scope_observer: None, + }, + ) + .await; + assert_eq!(result.status, ScannerCycleStatus::Incomplete); + + // Cycle 2: the checkpoint resumes under another plan and returns + // Partial. The bucket has no certified value, so the cycle stays + // incomplete and publishes no authoritative snapshot. + let (result, second) = run_cycle(&store, 2, complete_cycle(false, None)).await; + assert_eq!( + result.status, + ScannerCycleStatus::Incomplete, + "a bucket that never certified cannot be carried" + ); + assert_eq!(result.pending_dirty_usage_buckets(), 0); + assert!( + second.is_none_or(|snapshot| snapshot.usage_snapshot_observational || !snapshot.usage_snapshot_complete), + "an incomplete cycle publishes at most an observation" + ); + assert_eq!(dirty_usage_buckets_for_tests().len(), 1, "the bucket stays dirty"); + + // Cycle 3: the fresh sweep under one plan certifies the bucket. + let (result, third) = run_cycle(&store, 3, complete_cycle(false, None)).await; + let third = third.expect("a complete cycle publishes"); + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(third.usage_snapshot_converged, Some(true)); + assert_eq!(third.buckets_usage["fresh-bucket"].objects_count, 4); + assert!(result.clear_verified_usage().is_empty()); + assert!(dirty_usage_buckets_for_tests().is_empty()); + clear_dirty_usage_buckets_for_tests(); +} diff --git a/crates/scanner/src/scanner_io/tests/per_bucket_certification.rs b/crates/scanner/src/scanner_io/tests/per_bucket_certification.rs index 3b504c2e6..c3eca6776 100644 --- a/crates/scanner/src/scanner_io/tests/per_bucket_certification.rs +++ b/crates/scanner/src/scanner_io/tests/per_bucket_certification.rs @@ -4,7 +4,8 @@ //! Per-bucket certification on the single-disk fixture: a bucket written //! during the walk is reported pending while the other buckets are certified, //! the publication stays authoritative, and the next cycle is scoped to the -//! pending bucket. A guardrail veto still supersedes the whole cycle. +//! pending bucket. A PUT tail pending across the walk is ordinary namespace +//! activity; a guardrail veto still supersedes the whole cycle. use super::*; @@ -194,6 +195,61 @@ async fn write_during_walk_leaves_only_that_bucket_pending_and_scopes_the_next_c clear_dirty_usage_buckets_for_tests(); } +#[tokio::test] +#[serial] +async fn pending_put_tail_across_the_walk_no_longer_vetoes_the_cycle() { + let (_temp_dir, store) = setup_local_scanner_store(1, 1).await; + clear_dirty_usage_buckets_for_tests(); + for bucket in BUCKETS { + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + put_fixture_object(&store, bucket, "initial").await; + record_dirty_usage_bucket(bucket); + } + wait_for_namespace_commit_tails(store.as_ref()).await; + let vetoes_before = global_metrics().report().await.usage_freshness.guardrail_veto_total; + + // A PUT whose rename fanout is still pending when the post-walk probe + // runs used to flip the local publication barrier and supersede the + // whole cycle (rustfs/rustfs#8494). It is ordinary namespace activity: + // the walk certifies every bucket and publishes authority. + let pending_commit = ecstore_hold_namespace_commit(store.as_ref()); + assert!(ecstore_namespace_commits_pending(store.as_ref())); + assert!(!store.scanner_data_usage_publication_blocked().await); + let (result, snapshot) = run_cycle( + &store, + 1, + None, + true, + CycleHooks { + resolved_scope_observer: None, + post_walk_gate: None, + }, + ) + .await; + drop(pending_commit); + + assert_eq!(result.status, ScannerCycleStatus::Complete); + assert_eq!(result.pending_dirty_usage_buckets(), 0); + assert!(!snapshot.usage_snapshot_observational, "the publication takes the authoritative route"); + assert_eq!(snapshot.usage_snapshot_converged, Some(true)); + assert!(snapshot.usage_snapshot_pending_buckets.is_empty()); + assert!(snapshot.is_complete_bucket_usage_snapshot()); + for bucket in BUCKETS { + assert_eq!(snapshot.buckets_usage[bucket].objects_count, 1); + } + assert!(result.clear_verified_usage().is_empty()); + assert!(dirty_usage_buckets_for_tests().is_empty(), "a converged cycle clears the dirty map"); + let freshness = global_metrics().report().await.usage_freshness; + assert_eq!( + freshness.guardrail_veto_total, vetoes_before, + "a pending PUT tail is not a guardrail veto" + ); + clear_dirty_usage_buckets_for_tests(); +} + #[tokio::test] #[serial] async fn guardrail_veto_still_supersedes_the_cycle_onto_the_observed_route() { @@ -210,22 +266,29 @@ async fn guardrail_veto_still_supersedes_the_cycle_onto_the_observed_route() { wait_for_namespace_commit_tails(store.as_ref()).await; let vetoes_before = global_metrics().report().await.usage_freshness.guardrail_veto_total; - // A PUT tail held across the walk flips the local publication barrier, - // which is a guardrail veto regardless of the dirty map. - let pending_commit = ecstore_hold_namespace_commit(store.as_ref()); - assert!(store.scanner_data_usage_publication_blocked().await); + // A maintenance publication during the walk moves a guardrail field. It + // is not part of the coverage plan, so bucket checkpoints keep their + // plan, but the cycle it moved in is still superseded regardless of the + // dirty map: no cycle records whether it published that work. + let (observer, resolved_scope) = tokio::sync::oneshot::channel(); + let (release, gate) = tokio::sync::oneshot::channel(); + let maintainer = tokio::spawn(async move { + resolved_scope.await.expect("the cycle should report its resolved scope"); + record_scanner_maintenance_change("hot-bucket"); + release.send(()).expect("the cycle should still be waiting on the gate"); + }); let (result, snapshot) = run_cycle( &store, 1, None, true, CycleHooks { - resolved_scope_observer: None, - post_walk_gate: None, + resolved_scope_observer: Some(observer), + post_walk_gate: Some(gate), }, ) .await; - drop(pending_commit); + maintainer.await.expect("the maintenance task should finish"); assert_eq!(result.status, ScannerCycleStatus::Superseded); assert_eq!(result.pending_dirty_usage_buckets(), 0); @@ -239,7 +302,7 @@ async fn guardrail_veto_still_supersedes_the_cycle_onto_the_observed_route() { assert_eq!(freshness.guardrail_veto_total, vetoes_before + 1); assert_eq!( freshness.last_guardrail_veto_reason, - format!("publication_blocked@{}", crate::scanner::LOCAL_SCANNER_ACTIVITY_NODE) + format!("maintenance_generation@{}", crate::scanner::LOCAL_SCANNER_ACTIVITY_NODE) ); clear_dirty_usage_buckets_for_tests(); } diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index a333e0faa..ccfc8a60d 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -170,6 +170,7 @@ pub(crate) mod owner { #[cfg(test)] pub(crate) use rustfs_ecstore::api::set_disk::test_util::{ PutObjectCommitBarrier, PutObjectCommitPause, hold_namespace_commit as ecstore_hold_namespace_commit, + namespace_commits_pending as ecstore_namespace_commits_pending, }; pub(crate) use super::storage_contracts::{ diff --git a/docs/architecture/scanner-per-bucket-convergence.md b/docs/architecture/scanner-per-bucket-convergence.md index 100091248..99f57f829 100644 --- a/docs/architecture/scanner-per-bucket-convergence.md +++ b/docs/architecture/scanner-per-bucket-convergence.md @@ -1,7 +1,7 @@ # Scanner Per-Bucket Convergence Design **Use this when:** changing what makes a scanner cycle `Complete`, `Superseded`, or converged; adding or re-classifying a dirty-usage producer; deciding whether a background writeback may invalidate a scanner cycle; changing scanner worker concurrency or throttling; or activating scoped dirty acknowledgements. -**Source of truth:** `classify_nsscanner_cycle`, `scanner_cycle_activity_status`, and `scoped_scan_scope_from_dirty_buckets` in `crates/scanner/src/scanner_io.rs`; `dirty_usage_snapshot_status` and `acknowledge_scoped_dirty_usage` in `crates/scanner/src/scanner_io/dirty_usage.rs`; `scanner_concurrency_limit` in `crates/scanner/src/scanner_io/guards.rs`; the per-disk worker loop in `crates/scanner/src/scanner_io/io_cache.rs`; `DynamicSleeper` in `crates/scanner/src/sleeper.rs`; `ScannerNodeActivity` in `crates/scanner/src/scanner/activity.rs`; `persist_replication_state_if_current` in `crates/ecstore/src/bucket/replication/replication_resyncer.rs`; the publication fences in [scanner-usage-publication.md](scanner-usage-publication.md). +**Source of truth:** `classify_nsscanner_cycle`, `scanner_cycle_activity_status`, and `scoped_scan_scope_from_dirty_buckets` in `crates/scanner/src/scanner_io.rs`; `dirty_usage_snapshot_status` and `acknowledge_scoped_dirty_usage` in `crates/scanner/src/scanner_io/dirty_usage.rs`; `scanner_concurrency_limit` in `crates/scanner/src/scanner_io/guards.rs`; the per-disk worker loop and `carry_partial_bucket_values` in `crates/scanner/src/scanner_io/io_cache.rs`; `DynamicSleeper` in `crates/scanner/src/sleeper.rs`; `ScannerNodeActivity` in `crates/scanner/src/scanner/activity.rs`; `persist_replication_state_if_current` in `crates/ecstore/src/bucket/replication/replication_resyncer.rs`; the publication fences in [scanner-usage-publication.md](scanner-usage-publication.md). ## Decision @@ -63,8 +63,8 @@ the other: A third class sits beside these two and is easy to mistake for the first: - **Guardrail false positives**: the cycle is invalidated not by writes but - by the proof machinery itself. An in-flight PUT classified as data - movement, a journal lock timeout reported as an unavailable activity + by the proof machinery itself. An in-flight PUT counted as a publication + barrier, a journal lock timeout reported as an unavailable activity baseline, or one unreachable peer in a large cluster defers the whole cycle, and the dirty map is never cleared. Per-bucket certification does nothing for this class because the walk never reaches certification. The @@ -152,6 +152,17 @@ compare-and-remove semantics of `acknowledge_scoped_dirty_usage`: never move it; - generation moved by a `Namespace` hit: stays grey with the newer generation and counts as pending; +- walk stopped early (`Partial`): stays grey with its generation, like a + failed walk, and counts as pending; the value published for it is the + last certified one, never the partial walk's running total; +- a sweep that spans two cycles resumes only under the same coverage + identity (`scanner_activity_coverage_digest` with the bucket's own dirty + generation, through `prepare_bucket_checkpoint_with_coverage`). A peer's + namespace and dirty generations stay in that identity while its dirty + buckets are acknowledged per cycle: a write there between the two halves + of a sweep restarts the sweep instead of being acknowledged unwalked. The + local node's generations are left out, because its buckets are certified + one by one; a completed sweep is stored under the reuse identity; - a peer reported the bucket dirty without a scoped entry the coordinator can verify: stays grey. This is the mixed-version and partial-peer fallback, and it is per bucket, not per cycle. @@ -176,7 +187,20 @@ equality test to: - guardrail vetoed: `Superseded`; - otherwise: `Complete`, carrying `pending_dirty_buckets` as the count of - buckets left grey by `Namespace` hits. + buckets left grey by `Namespace` hits or by a walk that stopped early. + +A bucket walk that stops early without exhausting the cycle budget (a sweep +resumed under a newer plan, unresolved objects) is `Partial`. The set scan +stands in for it with the bucket's last certified value from the previous +complete set cache (`carry_partial_bucket_values` in +`crates/scanner/src/scanner_io/io_cache.rs`): usage is a sum, so the +certified buckets plus that value are a valid snapshot. The cycle then +completes with the bucket pending (`ScannerBucketScanStatus::PartialCarried`), +and the next scoped cycle walks it because the pending buckets of the +authoritative baseline join the dirty buckets in +`scoped_scan_scope_from_dirty_buckets`. Only a bucket with no certified +value on a set (its first walk, a changed incarnation, a hard failure) keeps +that set, and with it the cycle, `Incomplete`. `usage_snapshot_converged` becomes `pending_dirty_buckets == 0`. A `Complete` cycle with pending buckets publishes an authoritative snapshot: certified @@ -455,7 +479,8 @@ condition before merging. - `usage_freshness` in the admin scanner status reports `dirty_pending_buckets` (`Namespace` class only; after certification the - dirty map holds exactly the pending set), the attributed guardrail veto + dirty map holds the shaded pending buckets, and the published pending list + also names buckets whose walk stopped early), the attributed guardrail veto (`guardrail_veto_total`, `last_guardrail_veto_unix_secs`, `last_guardrail_veto_reason` as `@`), plus the pacer target, achieved rate, and whether a ceiling is binding. The console marks usage diff --git a/docs/architecture/scanner-usage-publication.md b/docs/architecture/scanner-usage-publication.md index e11810fbb..4985fc2ec 100644 --- a/docs/architecture/scanner-usage-publication.md +++ b/docs/architecture/scanner-usage-publication.md @@ -114,8 +114,15 @@ The structural scan-plan digest can remain stable across ordinary bucket writes so a scoped scan can retain unaffected baseline buckets. It is not sufficient proof for reusing a completed result within the same cycle. Bucket work uses an execution digest combining the structural plan and the full activity snapshot, -with the bucket's dirty generation included in its cache identity. Completed set -caches carry the same execution digest separately from their structural plan. +with the bucket's dirty generation included in its cache identity. A bucket +sweep's checkpoint is keyed separately by the coverage identity +(`scanner_activity_coverage_digest`, see +[scanner-per-bucket-convergence.md](scanner-per-bucket-convergence.md)), so +the reuse identity can change without restarting a sweep in progress. Completed +set caches carry the same execution digest separately from their structural +plan; a set cache that stands in for buckets whose walk stopped early binds +the carried bucket names into it (`scanner_carried_execution_digest`), so it +is never adopted in place of a full walk of the same cycle, nor the reverse. The persisted set-root fast path requires equal execution identities as well as the existing source, cycle, leader, tier, and cache-structure checks. @@ -289,9 +296,10 @@ worker finishing does not certify its namespace coverage. | `partial_saved_not_published` | The local or remote worker successfully saved a terminal Partial cache. Correlate its bucket, pool/set, cycle/leader epoch and `resume_position` with admission and sweep events. This confirms persistence, not complete coverage or authoritative publication. | | `checkpoint_identity_unavailable` | A local worker could not read the persisted bucket identity. `stage=admission` preserves the existing checkpoint without preparing a legacy rebuild; `stage=post_scan` rejects the unverified candidate before persistence. Correlate bucket, pool/set, cycle/epoch and the underlying read error. Remote terminal errors distinguish the same two stages. | | `checkpoint_identity_changed` | A successful post-scan read no longer matches the admitted identity, including a now-absent sidecar. Rejecting the result protects against bucket replacement; this differs from an unavailable read. | -| `remote_scan_partial` | The coordinator received a Partial terminal response for the named bucket and disk. Correlate the worker's bucket-sweep or checkpoint events; Partial is not a complete bucket root. | +| `remote_scan_partial` | The coordinator received a Partial terminal response for the named bucket and disk. Correlate the worker's bucket-sweep or checkpoint events; Partial is not a complete bucket root, but the set may still cover the bucket with its last certified value (`partial_buckets_carried`). | +| `partial_buckets_carried` | The set published the last certified value for `carried_buckets` buckets whose walk stopped early; `first_carried_bucket` names one. The cycle can complete with them pending and the next cycle walks them. A partial bucket without a certified value on the set is counted by `set_cache_publish_skipped` instead. | | `bucket_result_rejected` | A bucket result did not match the cycle's frozen tier-registry generation and was not applied. | -| `set_cache_publish_skipped` | `completed_buckets` counts distinct bucket roots successfully applied to this set cache. Use `first_incomplete_bucket` with pool/set/cycle to locate one actual gap in this set. Inspect `unprocessed_buckets` and budget/cancellation state, then correlate that bucket's worker events. | +| `set_cache_publish_skipped` | `completed_buckets` counts distinct bucket roots successfully applied to this set cache and `carried_buckets` the partial walks covered by a certified value. Use `first_incomplete_bucket` with pool/set/cycle to locate one actual gap in this set. Inspect `unprocessed_buckets` and budget/cancellation state, then correlate that bucket's worker events. | | `set_scan_completed` | Set workers have finished. `coverage_complete` reports the structural coverage gate, not durable publication. This event alone does not prove a complete set snapshot. | | `incomplete_coverage` | The cycle did not obtain complete structural coverage. This does not identify a hot bucket or prove that ordinary writes caused the gap. | | `publication_proof_pending` | Publication is waiting for a retryable proof. Use cycle/stage, `pending_duration`, the preceding `publication_proof_started` kind, and the peer/purpose in the error. This event does not assert that the walk was Complete. | @@ -347,9 +355,10 @@ inventories. Keep logs access-controlled and redact deployment identifiers before sharing them. An unlimited walk budget can also leave a retryable authoritative publication -waiting for leadership loss or a successful proof. Observational publication -still retains the movement fence, while authoritative publication requires the -strict namespace proof. Do not remove those fences to clear a status. Verify +waiting for leadership loss or a successful proof. Observational and +authoritative publication both retain the movement fence, on this node and +through the remote publication leases. Do not remove those fences to clear a +status. Verify that a deployment contains the relevant fixes by commit ancestry; a newer release name alone is not evidence that it includes current main. diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index e92c637ad..d28b8f581 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -6689,7 +6689,7 @@ mod tests { "the standard builder must use this exact instance context" ); super::timeout(Duration::from_secs(10), async { - while env.ecstore.scanner_data_usage_publication_blocked().await { + while env.ecstore.namespace_commits_pending_for_test() { tokio::task::yield_now().await; } }) @@ -7715,7 +7715,7 @@ mod tests { before ); assert!(!disk.path().join("target-rpc/destination").exists()); - assert!(!fixture.env.ecstore.scanner_data_usage_publication_blocked().await); + assert!(!fixture.env.ecstore.namespace_commits_pending_for_test()); } #[tokio::test] @@ -7748,7 +7748,7 @@ mod tests { .expect("source remains"), before ); - assert!(!fixture.env.ecstore.scanner_data_usage_publication_blocked().await); + assert!(!fixture.env.ecstore.namespace_commits_pending_for_test()); } #[cfg(not(windows))] @@ -7975,12 +7975,12 @@ mod tests { .expect("bounded physical entry"); drop(handler); assert!( - fixture.env.ecstore.scanner_data_usage_publication_blocked().await, + fixture.env.ecstore.namespace_commits_pending_for_test(), "dropping the actual target handler must not release its physical owner" ); drop(pause); super::timeout(Duration::from_secs(10), async { - while fixture.env.ecstore.scanner_data_usage_publication_blocked().await { + while fixture.env.ecstore.namespace_commits_pending_for_test() { tokio::task::yield_now().await; } }) @@ -8061,10 +8061,10 @@ mod tests { .await .expect("bounded physical restore entry"); drop(handler); - assert!(fixture.env.ecstore.scanner_data_usage_publication_blocked().await); + assert!(fixture.env.ecstore.namespace_commits_pending_for_test()); drop(pause); super::timeout(Duration::from_secs(10), async { - while fixture.env.ecstore.scanner_data_usage_publication_blocked().await { + while fixture.env.ecstore.namespace_commits_pending_for_test() { tokio::task::yield_now().await; } }) @@ -8208,7 +8208,7 @@ mod tests { // Copy the complete internal trees, including erasure part data, // without editing disk IDs, cluster identity, epochs or pool topology. super::timeout(Duration::from_secs(10), async { - while store_b.scanner_data_usage_publication_blocked().await { + while store_b.namespace_commits_pending_for_test() { tokio::task::yield_now().await; } }) @@ -8231,7 +8231,7 @@ mod tests { } assert_eq!(internal_snapshot(root_a.path()).await, snapshot, "A must receive the complete physical snapshot"); assert_eq!(internal_snapshot(root_b.path()).await, snapshot, "B's source snapshot must remain unchanged"); - assert!(!store_b.scanner_data_usage_publication_blocked().await); + assert!(!store_b.namespace_commits_pending_for_test()); assert_eq!(store_b.scanner_namespace_mutation_generation(), snapshot_generation); let store_a = build_store(root_a.path()).await; assert_eq!( @@ -8361,8 +8361,8 @@ mod tests { assert!(tokio::fs::try_exists(&a_staging).await.expect("A staging exists")); assert_ne!(a_before, b_before, "the old on-disk bodies must distinguish A from B"); super::timeout(Duration::from_secs(10), async { - while store_a.scanner_data_usage_publication_blocked().await - || store_b.scanner_data_usage_publication_blocked().await + while store_a.namespace_commits_pending_for_test() + || store_b.namespace_commits_pending_for_test() { tokio::task::yield_now().await; } @@ -8411,8 +8411,8 @@ mod tests { store_b.scanner_namespace_mutation_generation(), ); let pending_after = ( - store_a.scanner_data_usage_publication_blocked().await, - store_b.scanner_data_usage_publication_blocked().await, + store_a.namespace_commits_pending_for_test(), + store_b.namespace_commits_pending_for_test(), ); for disk in store_a.disk_map.values().chain(store_b.disk_map.values()).flatten().flatten() { disk.close().await.expect("close real fixture disk before assertions and directory cleanup");