gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-26 22:07:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): stop treating an ordinary PUT as data movement in cycle termination (#8696)

Browse Source 
* fix(scanner): stop treating an ordinary PUT as data movement in cycle termination

A continuously written deployment never reached a `Complete` scanner cycle
(#8494, #8305, #7108). Three places in the termination path vetoed the cycle
for an ordinary PUT; each now lets the write shade its bucket, the path
per-bucket certification (#8542) built, and keeps the data-movement fences.

1. Usage publication is fenced by data movement only.
   `scanner_data_usage_publication_blocked` no longer includes
   `namespace_commits_pending()`, so the post-walk probe, the local defer
   decision, the post-lease recheck and the usage-store route probes stop
   superseding or deferring a cycle for a pending rename fanout. The scanner
   publication leases (acquire, validate, target guard) drop the
   pending-commit rejection and the namespace-generation staleness check and
   keep the movement generation, the operation gate, generation exhaustion
   and the purpose/target restriction. A test-util probe lets fixtures wait
   for tails directly.

2. The plan identity is split into a coverage identity and a reuse identity.
   `scanner_activity_coverage_digest` (peer set, protocol, topology, movement,
   and a remote peer's generations) is what a bucket checkpoint is started
   and resumed under, through `ScannerBucketScanPlan::checkpoint_coverage_digest`,
   `DataUsageCacheReuseOptions::checkpoint_plan_digest` and
   `prepare_bucket_checkpoint_with_coverage`. The reuse identity
   (`scanner_activity_structural_digest`) keeps `instance_id` and
   `maintenance_generation`, so a restart or a maintenance change still
   invalidates the baseline and the set caches; a completed sweep is stored
   under it. `publication_blocked` leaves every digest and stays a guardrail
   veto. A remote worker still resumes under the single digest on the wire.

3. A bucket walk that stops early no longer makes the cycle `Incomplete`
   when the bucket has a certified value. The set scan stands in with the
   bucket's flattened value from its previous complete set cache
   (`carry_partial_bucket_values`), the cycle reports it pending
   (`ScannerBucketScanStatus::PartialCarried`), keeps it in the dirty map,
   and the next scoped cycle walks it because the baseline's pending list
   joins the dirty buckets. A bucket with no certified value keeps the cycle
   `Incomplete`; failed buckets are unchanged.

No persisted file, generation, receipt or environment variable is added.
Refs #8494, #8305, #7108, rustfs/backlog#2903.

* chore(ecstore): gate the namespace commit probe to test builds

The scanner publication barrier no longer consults
InstanceContext::namespace_commits_pending, so it has no production
caller left; keep it for tests and the test-util facade only.

* test(rustfs): observe namespace commit ownership directly in node_service tests

The scanner usage publication barrier is movement-only now, so it no
longer reports a pending rename. The node_service tests that assert a
cancelled handler keeps its physical owner, or that wait for rename
tails to drain, read the namespace commit state through a test-only
ECStore probe instead.
This commit is contained in:
Chris
2026-10-19 22:23:59 +09:00
committed by GitHub
parent c48ff0bbbd
commit 7c75f319bc
26 changed files with 1467 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/api/mod.rs
+8
View File
@@ -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()
}
}
}
crates/ecstore/src/runtime/instance.rs
+4
View File
@@ -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
}
crates/ecstore/src/store/init.rs
+21 -24
View File
@@ -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]
crates/ecstore/src/store/mod.rs
+63 -56
View File
@@ -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());
crates/ecstore/src/store/peer.rs
+25 -22
View File
@@ -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)
crates/scanner/src/data_usage_define.rs
+33 -3
View File
@@ -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;
crates/scanner/src/remote_scanner/stream.rs
+3
View File
@@ -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 {
crates/scanner/src/scanner.rs
+23 -28
View File
@@ -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<S>(storeapi: &S, observational_only: bool) -> Option<ScannerCycleDeferReason>
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::{
crates/scanner/src/scanner/activity.rs
+54 -8
View File
@@ -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,
crates/scanner/src/scanner/tests.rs
+205 -35
View File
@@ -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<dyn Fn(&mut ScannerNodeActivity)>);
// 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<Mutation> = 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<Mutation> = 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<Mutation> = 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([
crates/scanner/src/scanner/tests/scoped_ack_publication.rs
+1 -1
View File
@@ -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;
}
})
crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs
+3
View File
@@ -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;
crates/scanner/src/scanner_io.rs
+70 -16
View File
@@ -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<String> {
proof
.authoritative_data
.and_then(|data| serde_json::from_slice::<DataUsageInfo>(data).ok())
.map(|snapshot| snapshot.usage_snapshot_pending_buckets)
.unwrap_or_default()
}
fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option<DataUsageScanPlanDigest> {
let authoritative = serde_json::from_slice::<DataUsageInfo>(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::<HashSet<_>>();
let selected_buckets = dirty_buckets
.into_iter()
.chain(scanner_baseline_pending_buckets(&baseline_proof))
.filter(|bucket| current_buckets.contains(bucket.as_str()))
.collect::<HashSet<_>>();
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<Arc<StdMutex<ScannerServiceCohort>>>,
// 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<Mutex<HashSet<String>>>,
partial: Arc<Mutex<HashSet<String>>>,
/// 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<Mutex<HashSet<String>>>,
namespace_not_found: Arc<Mutex<HashSet<String>>>,
}
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::<Vec<_>>();
buckets.sort_unstable_by(|left, right| left.name.cmp(&right.name));
@@ -953,17 +983,31 @@ fn finalize_nsscanner_result(results: &[DataUsageCache], first_err: Option<Error
enum ScannerBucketScanStatus {
Complete,
Failed,
/// At least one bucket walk stopped early and no certified value could
/// stand in for it on some set.
Partial,
/// Every bucket walk that stopped early was covered by that bucket's
/// last certified set value, so the cycle may complete with those
/// buckets pending. Necessary, not sufficient: `completed_all_sets`
/// still proves that every set published.
PartialCarried,
NamespaceNotFound,
}
fn scanner_bucket_scan_status(has_failed: bool, has_partial: bool, has_namespace_not_found: bool) -> 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<S>(
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) => {
crates/scanner/src/scanner_io/cache.rs
+41 -5
View File
@@ -286,6 +286,10 @@ pub(crate) struct DataUsageCacheReuseOptions {
pub(crate) require_source: bool,
pub(crate) tier_registry_generation: Option<u64>,
pub(crate) checkpoint_identity: Option<crate::DataUsageScanIdentity>,
/// 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<DataUsageScanPlanDigest>,
}
#[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(
crates/scanner/src/scanner_io/dirty_usage.rs
+3 -1
View File
@@ -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<String> {
let dirty_buckets = dirty_usage_buckets();
dirty_buckets
crates/scanner/src/scanner_io/io_cache.rs
+143 -8
View File
@@ -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<String>,
bucket_failures: &ScannerBucketFailureState,
current_bucket_incarnations: Option<&HashMap<String, uuid::Uuid>>,
) -> Vec<String> {
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(),
crates/scanner/src/scanner_io/io_cycle.rs
+18 -4
View File
@@ -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::<HashSet<_>>();
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 {
crates/scanner/src/scanner_io/publish_gate_tests.rs
+1
View File
@@ -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,
},
);
crates/scanner/src/scanner_io/tests.rs
+175 -17
View File
@@ -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]
crates/scanner/src/scanner_io/tests/checkpoint_coverage_identity.rs
+171
View File
@@ -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<ECStore>, 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<ECStore>,
cycle: u64,
max_objects: Option<u64>,
) -> (ScannerCycleResult, Option<DataUsageInfo>, Arc<ScannerCycleBudget>) {
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();
}
crates/scanner/src/scanner_io/tests/partial_bucket_carry.rs
+270
View File
@@ -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<ECStore>, 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<Bytes>,
requires_full_scan: bool,
max_objects: Option<u64>,
resolved_scope_observer: Option<tokio::sync::oneshot::Sender<ScannerBucketScanScope>>,
}
/// Run one cycle; an incomplete cycle may publish nothing or an observation.
async fn run_cycle(store: &Arc<ECStore>, cycle: u64, options: CycleOptions) -> (ScannerCycleResult, Option<DataUsageInfo>) {
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<_>>(),
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();
}
crates/scanner/src/scanner_io/tests/per_bucket_certification.rs
+72 -9
View File
@@ -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();
}
crates/scanner/src/storage_api.rs
+1
View File
@@ -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::{
docs/architecture/scanner-per-bucket-convergence.md
+30 -5
View File
@@ -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 `<field>@<host>`), plus the pacer target,
achieved rate, and whether a ceiling is binding. The console marks usage
docs/architecture/scanner-usage-publication.md
+16 -7
View File
@@ -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.
rustfs/src/storage/rpc/node_service.rs
+13 -13
View File
@@ -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");