gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-01 17:58:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): preserve cache cycle during usage recovery (#6921)

Browse Source 
Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-31 08:19:54 +08:00
committed by GitHub
parent e6234d3714
commit 8ecd8f2520
4 changed files with 468 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/scanner.rs
+61 -12
View File
@@ -751,15 +751,15 @@ fn prepare_cycle_for_usage_floor_bootstrap(
cycle_info: &mut CurrentCycle,
usage_floor: PersistedUsageFloor,
startup: PersistedUsageFloorStartup,
) -> (bool, bool) {
) -> (bool, ScannerCycleResetPolicy) {
match startup {
PersistedUsageFloorStartup::Authoritative => (false, false),
PersistedUsageFloorStartup::Authoritative => (false, ScannerCycleResetPolicy::None),
PersistedUsageFloorStartup::Missing => {
// Cycle progress without its corresponding usage floor cannot
// prove namespace coverage. Restart from cycle zero while keeping
// the separately fenced leader epoch monotonic.
*cycle_info = CurrentCycle::default();
(true, true)
(true, ScannerCycleResetPolicy::ResetAll)
}
PersistedUsageFloorStartup::BootstrapPending => {
// An unfenced marker may have been written before an upgrade's old
@@ -768,14 +768,25 @@ fn prepare_cycle_for_usage_floor_bootstrap(
if usage_floor.leader_epoch == 0 {
*cycle_info = CurrentCycle::default();
}
(true, usage_floor.leader_epoch == 0)
(
true,
if usage_floor.leader_epoch == 0 {
ScannerCycleResetPolicy::ResetAll
} else {
ScannerCycleResetPolicy::None
},
)
}
PersistedUsageFloorStartup::RecoveredLegacyEmptyFence => {
// The legacy empty fence proves only its leader epoch, not
// namespace coverage. Restart coverage from zero while retaining
// that epoch as the lower bound for the next leadership claim.
*cycle_info = CurrentCycle::default();
(true, true)
// namespace coverage. Clear coverage while retaining the durable
// cycle number so surviving caches cannot force a regression.
let next = cycle_info.next;
*cycle_info = CurrentCycle {
next,
..Default::default()
};
(true, ScannerCycleResetPolicy::ResetCoveragePreservingNext)
}
}
}
@@ -1296,7 +1307,15 @@ where
LockLost: Future<Output = ()>,
{
let fence_ctx = ctx.child_token();
let claim = claim_scanner_leadership(&fence_ctx, storeapi, cycle_info, cycle_revision, leader_epoch, false, false);
let claim = claim_scanner_leadership(
&fence_ctx,
storeapi,
cycle_info,
cycle_revision,
leader_epoch,
false,
ScannerCycleResetPolicy::None,
);
tokio::pin!(claim);
tokio::pin!(lock_lost);
tokio::select! {
@@ -1792,8 +1811,18 @@ async fn run_data_scanner_cycle_with_budget(
mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
return ScannerCycleOutcome::Failed;
}
match scanner_cycle_pre_commit_outcome(scan_cycle_result.required_cycle_floor(), &usage_persist_outcome) {
let required_cycle_floor = scan_cycle_result.required_cycle_floor();
let pre_commit_outcome = scanner_cycle_pre_commit_outcome(required_cycle_floor, &usage_persist_outcome);
update_scanner_cache_cycle_recovery_status(
cycle_info.current,
leader_epoch,
required_cycle_floor,
pre_commit_outcome,
scan_cycle_result.status == ScannerCycleStatus::Complete,
);
match pre_commit_outcome {
Some(ScannerCyclePreCommitOutcome::RecoverCacheCycle(required_cycle)) => {
record_scanner_cache_cycle_recovery_attempt();
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
@@ -2312,7 +2341,7 @@ async fn run_data_scanner_with_maintenance_state(
return Ok(());
}
};
let (allow_usage_floor_bootstrap_pending, reset_usage_floor_bootstrap_cycle_on_conflict) =
let (allow_usage_floor_bootstrap_pending, usage_floor_cycle_reset_policy) =
prepare_cycle_for_usage_floor_bootstrap(&mut cycle_info, usage_floor, usage_floor_startup);
apply_persisted_usage_floor(&mut cycle_info, &mut leader_epoch, usage_floor);
match usage_floor_startup {
@@ -2374,7 +2403,7 @@ async fn run_data_scanner_with_maintenance_state(
&mut cycle_revision,
&mut leader_epoch,
allow_usage_floor_bootstrap_pending,
reset_usage_floor_bootstrap_cycle_on_conflict,
usage_floor_cycle_reset_policy,
),
guard.lock_lost_notified(),
)
@@ -2902,6 +2931,26 @@ fn scanner_cycle_pre_commit_outcome(
}
}
fn update_scanner_cache_cycle_recovery_status(
requested_cycle: u64,
leader_epoch: u64,
required_cycle_floor: Option<u64>,
pre_commit_outcome: Option<ScannerCyclePreCommitOutcome>,
cache_scope_complete: bool,
) {
match (required_cycle_floor, pre_commit_outcome) {
(Some(required_cycle), _) => {
record_scanner_cache_cycle_ahead(requested_cycle, required_cycle, leader_epoch);
}
(None, Some(ScannerCyclePreCommitOutcome::Deferred(_))) => {
// A deferred scan may not have covered the cache that established
// the existing floor, so it cannot prove recovery is complete.
}
(None, _) if cache_scope_complete => clear_scanner_cache_cycle_ahead(),
(None, _) => {}
}
}
fn scanner_cycle_completion_outcome(
scan_status: ScannerCycleStatus,
usage_persist_outcome: DataUsagePersistOutcome,
crates/scanner/src/scanner/cycle_state.rs
+53 -2
View File
@@ -25,6 +25,7 @@ const METRIC_SCANNER_CYCLE_RECOVERY_REQUIRED: &str = "rustfs_scanner_cycle_recov
const METRIC_SCANNER_CYCLE_RECOVERY_RETRY_COUNT: &str = "rustfs_scanner_cycle_recovery_retry_count";
const USAGE_FLOOR_LOAD_FAILED: &str = "usage_floor_load_failed";
const LEGACY_EMPTY_USAGE_FLOOR_RECOVERY: &str = "legacy_empty_usage_floor";
const CACHE_CYCLE_AHEAD: &str = "cache_cycle_ahead";
#[derive(Clone, Debug, Default, Serialize)]
pub struct ScannerCycleRecoveryStatus {
@@ -40,6 +41,7 @@ pub struct ScannerCycleRecoveryStatus {
pub first_detected_at_unix_secs: Option<u64>,
pub last_attempt_at_unix_secs: Option<u64>,
pub retry_count: u64,
/// Maximum automatic retries, or zero when the recovery is unbounded.
pub max_retries: u32,
/// Whether the scanner may retry this state automatically.
pub retryable: bool,
@@ -72,6 +74,7 @@ fn set_scanner_cycle_recovery_status(status: ScannerCycleRecoveryStatus) {
| "cleanup-pending"
| "usage_floor_load_failed"
| "usage_floor_recovery_pending"
| "cache_cycle_ahead"
) {
1.0
} else {
@@ -113,6 +116,51 @@ pub(super) fn clear_scanner_usage_floor_failure() {
}
}
pub(super) fn record_scanner_cache_cycle_ahead(requested_cycle: u64, required_cycle: u64, leader_epoch: u64) {
let previous = scanner_cycle_recovery_status();
let same_floor = previous.classification.as_deref() == Some(CACHE_CYCLE_AHEAD)
&& previous.generation == Some(required_cycle)
&& previous.leader_epoch == Some(leader_epoch);
let now = unix_now_secs();
let (first_detected_at_unix_secs, retry_count) = if same_floor {
(previous.first_detected_at_unix_secs.or(Some(now)), previous.retry_count)
} else {
(Some(now), 0)
};
set_scanner_cycle_recovery_status(ScannerCycleRecoveryStatus {
path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
state: CACHE_CYCLE_AHEAD.to_string(),
classification: Some(CACHE_CYCLE_AHEAD.to_string()),
generation: Some(required_cycle),
leader_epoch: Some(leader_epoch),
first_detected_at_unix_secs,
last_attempt_at_unix_secs: Some(now),
retry_count,
max_retries: 0,
retryable: true,
reason: Some(format!(
"persisted scanner cache cycle {required_cycle} is ahead of requested cycle {requested_cycle}"
)),
..Default::default()
});
}
pub(super) fn record_scanner_cache_cycle_recovery_attempt() {
let mut status = scanner_cycle_recovery_status();
if status.classification.as_deref() != Some(CACHE_CYCLE_AHEAD) {
return;
}
status.retry_count = status.retry_count.saturating_add(1);
status.last_attempt_at_unix_secs = Some(unix_now_secs());
set_scanner_cycle_recovery_status(status);
}
pub(super) fn clear_scanner_cache_cycle_ahead() {
if scanner_cycle_recovery_status().classification.as_deref() == Some(CACHE_CYCLE_AHEAD) {
set_scanner_cycle_recovery_status(recovery_status("healthy", None, false));
}
}
pub(super) fn record_legacy_empty_usage_floor_recovery_pending(leader_epoch: u64) {
let previous = scanner_cycle_recovery_status();
let same_recovery = previous.classification.as_deref() == Some(LEGACY_EMPTY_USAGE_FLOOR_RECOVERY)
@@ -147,9 +195,12 @@ pub(super) fn clear_legacy_empty_usage_floor_recovery_status() {
pub(super) fn record_scanner_cycle_recovery_retry(attempt: u32) -> bool {
let mut status = scanner_cycle_recovery_status();
status.retry_count = u64::from(attempt);
if status.classification.as_deref() == Some(CACHE_CYCLE_AHEAD) {
return true;
}
status.retry_count = status.retry_count.max(u64::from(attempt));
status.last_attempt_at_unix_secs = Some(unix_now_secs());
if attempt >= MAX_SCANNER_CYCLE_RECOVERY_RETRIES {
if status.max_retries != 0 && status.retry_count >= u64::from(status.max_retries) {
status.state = "paused".to_string();
status.retryable = false;
status.reason = Some("scanner cycle recovery retry budget reached; sparse backend probes continue".to_string());
crates/scanner/src/scanner/leadership.rs
+28 -10
View File
@@ -21,6 +21,29 @@ pub(super) enum ScannerLeadershipClaimReconcile {
Unchanged,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum ScannerCycleResetPolicy {
None,
ResetAll,
ResetCoveragePreservingNext,
}
impl ScannerCycleResetPolicy {
fn apply(self, cycle_info: &mut CurrentCycle, attempted_next: u64) {
match self {
Self::None => {}
Self::ResetAll => *cycle_info = CurrentCycle::default(),
Self::ResetCoveragePreservingNext => {
let next = cycle_info.next.max(attempted_next);
*cycle_info = CurrentCycle {
next,
..Default::default()
};
}
}
}
}
pub(super) async fn reconcile_scanner_leadership_claim(
storeapi: Arc<impl ScannerObjectIO>,
attempted: &[u8],
@@ -329,7 +352,7 @@ pub(super) async fn claim_scanner_leadership(
revision: &mut DataUsageCacheRevision,
persisted_epoch: &mut u64,
allow_bootstrap_pending: bool,
reset_bootstrap_cycle_on_conflict: bool,
cycle_reset_policy: ScannerCycleResetPolicy,
) -> bool {
for retry in 0..=SCANNER_PERSIST_CAS_RETRIES {
if ctx.is_cancelled() {
@@ -347,6 +370,7 @@ pub(super) async fn claim_scanner_leadership(
);
return false;
};
let attempted_next = cycle_info.next;
let data = match encode_scanner_cycle_state(cycle_info, claimed_epoch) {
Ok(data) => data,
Err(err) => {
@@ -459,9 +483,7 @@ pub(super) async fn claim_scanner_leadership(
.await;
}
Ok(ScannerLeadershipClaimReconcile::Changed) if retry < SCANNER_PERSIST_CAS_RETRIES => {
if reset_bootstrap_cycle_on_conflict {
*cycle_info = CurrentCycle::default();
}
cycle_reset_policy.apply(cycle_info, attempted_next);
continue;
}
Ok(ScannerLeadershipClaimReconcile::Changed | ScannerLeadershipClaimReconcile::Unchanged) => {
@@ -517,17 +539,13 @@ pub(super) async fn claim_scanner_leadership(
Ok(ScannerLeadershipClaimReconcile::Changed)
if retry < SCANNER_PERSIST_CAS_RETRIES && !ctx.is_cancelled() =>
{
if reset_bootstrap_cycle_on_conflict {
*cycle_info = CurrentCycle::default();
}
cycle_reset_policy.apply(cycle_info, attempted_next);
continue;
}
Ok(ScannerLeadershipClaimReconcile::Unchanged)
if precondition_failed && retry < SCANNER_PERSIST_CAS_RETRIES && !ctx.is_cancelled() =>
{
if reset_bootstrap_cycle_on_conflict {
*cycle_info = CurrentCycle::default();
}
cycle_reset_policy.apply(cycle_info, attempted_next);
continue;
}
Ok(ScannerLeadershipClaimReconcile::Changed | ScannerLeadershipClaimReconcile::Unchanged) => {
crates/scanner/src/scanner/tests.rs
+326 -30
View File
@@ -16,10 +16,11 @@ use super::heal_info::{classify_background_heal_read_error, decode_background_he
use super::*;
use crate::EcstoreResult;
use crate::{
DATA_USAGE_BLOOM_RECOVERY_PATH, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints,
ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions,
ScannerPutObjReader as PutObjReader, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests,
init_local_disks_with_instance_ctx,
DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_CACHE_KEY_FORMAT, DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT,
DataUsageCachePrepareOutcome, DataUsageCacheSource, DataUsageEntry, DataUsageScanPlanDigest, Endpoint, EndpointServerPools,
Endpoints, InstanceContext, PoolEndpoints, ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo,
ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader, init_bucket_metadata_sys_for_scanner_tests,
init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx,
};
use serial_test::serial;
use std::collections::{HashMap, HashSet};
@@ -1991,7 +1992,7 @@ fn rc3_legacy_empty_usage_fence(epoch: Option<u64>) -> Vec<u8> {
}
#[tokio::test]
async fn scanner_usage_floor_recovers_rc3_empty_fences_and_restarts_from_zero() {
async fn scanner_usage_floor_recovers_rc3_empty_fences_and_preserves_cycle_number() {
let store = Arc::new(MemoryConfigStore::default());
let primary_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
@@ -2034,16 +2035,17 @@ async fn scanner_usage_floor_recovers_rc3_empty_fences_and_restarts_from_zero()
assert_eq!(restart_floor.leader_epoch, 7);
assert_eq!(restart_state, PersistedUsageFloorStartup::RecoveredLegacyEmptyFence);
let mut cycle = CurrentCycle {
current: 41,
next: 42,
..Default::default()
current: 17_117,
next: 17_118,
cycle_completed: vec![Utc::now()],
started: Utc::now(),
};
assert_eq!(
prepare_cycle_for_usage_floor_bootstrap(&mut cycle, restart_floor, restart_state),
(true, true)
(true, ScannerCycleResetPolicy::ResetCoveragePreservingNext)
);
assert_eq!(cycle.current, 0);
assert_eq!(cycle.next, 0);
assert_eq!(cycle.next, 17_118);
assert!(cycle.cycle_completed.is_empty());
let mut revision = DataUsageCacheRevision::Missing;
@@ -2056,11 +2058,70 @@ async fn scanner_usage_floor_recovers_rc3_empty_fences_and_restarts_from_zero()
&mut revision,
&mut leader_epoch,
true,
true,
ScannerCycleResetPolicy::ResetCoveragePreservingNext,
)
.await
);
assert_eq!(leader_epoch, 8);
assert_eq!(cycle.next, 17_118);
assert_eq!(cycle.current, 0);
assert!(cycle.cycle_completed.is_empty());
let source = DataUsageCacheSource::new(0, 0);
let scan_plan_digest = DataUsageScanPlanDigest([7; 32]);
for (cache_path, name) in [
(DATA_USAGE_CACHE_NAME.to_string(), DATA_USAGE_ROOT),
(format!("photos/{DATA_USAGE_CACHE_NAME}"), "photos"),
] {
let mut historical = DataUsageCache::default();
historical.info.name = name.to_string();
historical.info.next_cycle = 17_118;
historical.info.leader_epoch = 7;
historical.info.source = Some(source);
historical.info.scan_plan_digest = Some(scan_plan_digest);
historical.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT;
historical.info.snapshot_complete = true;
historical.replace(name, "", DataUsageEntry::default());
historical
.save(store.clone(), &cache_path)
.await
.expect("historical scanner cache should persist through the storage path");
let mut recovered = DataUsageCache::default();
let revisions = recovered
.load_with_revisions(store.clone(), &cache_path)
.await
.expect("historical scanner cache should reload with CAS revisions");
assert_eq!(recovered.info.name, name);
assert_eq!(recovered.info.next_cycle, 17_118);
assert_eq!(recovered.info.leader_epoch, 7);
assert!(!recovered.cache.is_empty());
assert_eq!(
recovered.prepare_for_scan(name, cycle.next, leader_epoch, source, scan_plan_digest, true),
DataUsageCachePrepareOutcome::Reset,
"recovered cache should reset without a cycle regression: {cache_path}"
);
assert_eq!(recovered.info.next_cycle, 17_118);
assert_eq!(recovered.info.leader_epoch, 8);
assert!(!recovered.info.snapshot_complete);
assert!(recovered.cache.is_empty());
recovered
.save_with_revisions(store.clone(), &cache_path, &revisions)
.await
.expect("reset scanner cache should persist with its loaded revisions");
let mut persisted_reset = DataUsageCache::default();
persisted_reset
.load(store.clone(), &cache_path)
.await
.expect("persisted reset scanner cache should reload");
assert_eq!(persisted_reset.info.name, name);
assert_eq!(persisted_reset.info.next_cycle, 17_118);
assert_eq!(persisted_reset.info.leader_epoch, 8);
assert!(!persisted_reset.info.snapshot_complete);
assert!(persisted_reset.cache.is_empty());
}
complete_legacy_empty_usage_floor_recovery(store.clone(), leader_epoch)
.await
.expect("leadership claim should retire the recovery marker");
@@ -2253,7 +2314,7 @@ async fn scanner_usage_floor_recovery_reconciles_marker_delete_post_commit_error
let mut cycle = CurrentCycle::default();
let mut revision = DataUsageCacheRevision::Missing;
let mut leader_epoch = floor.leader_epoch;
let (allow_pending, reset_on_conflict) = prepare_cycle_for_usage_floor_bootstrap(&mut cycle, floor, state);
let (allow_pending, cycle_reset_policy) = prepare_cycle_for_usage_floor_bootstrap(&mut cycle, floor, state);
assert!(
claim_scanner_leadership(
&CancellationToken::new(),
@@ -2262,7 +2323,7 @@ async fn scanner_usage_floor_recovery_reconciles_marker_delete_post_commit_error
&mut revision,
&mut leader_epoch,
allow_pending,
reset_on_conflict,
cycle_reset_policy,
)
.await
);
@@ -2494,6 +2555,53 @@ fn scanner_usage_floor_recovery_stays_retryable_until_claim_cleanup() {
assert_eq!(scanner_cycle_recovery_status().state, "healthy");
}
#[test]
#[serial]
fn scanner_cache_cycle_ahead_is_visible_until_a_later_scan_clears_it() {
record_scanner_cache_cycle_ahead(0, 17_118, 8);
let pending = scanner_cycle_recovery_status();
assert_eq!(pending.state, "cache_cycle_ahead");
assert_eq!(pending.classification.as_deref(), Some("cache_cycle_ahead"));
assert_eq!(pending.generation, Some(17_118));
assert_eq!(pending.leader_epoch, Some(8));
assert!(pending.retryable);
assert_eq!(pending.max_retries, 0);
assert_eq!(
pending.reason.as_deref(),
Some("persisted scanner cache cycle 17118 is ahead of requested cycle 0")
);
let first_detected = pending.first_detected_at_unix_secs;
record_scanner_cache_cycle_ahead(0, 17_118, 8);
let observed_again = scanner_cycle_recovery_status();
assert_eq!(observed_again.retry_count, 0);
assert_eq!(observed_again.first_detected_at_unix_secs, first_detected);
assert!(record_scanner_cycle_recovery_retry(4));
assert_eq!(scanner_cycle_recovery_status().retry_count, 0);
record_scanner_cache_cycle_recovery_attempt();
record_scanner_cache_cycle_recovery_attempt();
let retried = scanner_cycle_recovery_status();
assert_eq!(retried.retry_count, 2);
assert!(retried.retryable);
update_scanner_cache_cycle_recovery_status(
0,
8,
None,
Some(ScannerCyclePreCommitOutcome::Deferred(ScannerCycleDeferReason::DataMovement)),
false,
);
assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("cache_cycle_ahead"));
update_scanner_cache_cycle_recovery_status(17_118, 8, None, None, false);
assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("cache_cycle_ahead"));
update_scanner_cache_cycle_recovery_status(17_118, 8, None, None, true);
assert_eq!(scanner_cycle_recovery_status().state, "healthy");
}
#[tokio::test]
#[serial]
async fn scanner_usage_floor_failure_clears_stale_leader_liveness() {
@@ -2946,7 +3054,7 @@ fn missing_usage_floor_discards_unfenced_cycle_progress() {
assert_eq!(
prepare_cycle_for_usage_floor_bootstrap(&mut cycle, PersistedUsageFloor::default(), PersistedUsageFloorStartup::Missing,),
(true, true)
(true, ScannerCycleResetPolicy::ResetAll)
);
assert_eq!(cycle.next, 0);
assert_eq!(cycle.current, 0);
@@ -2959,7 +3067,7 @@ fn missing_usage_floor_discards_unfenced_cycle_progress() {
PersistedUsageFloor::default(),
PersistedUsageFloorStartup::BootstrapPending,
),
(true, true)
(true, ScannerCycleResetPolicy::ResetAll)
);
assert_eq!(cycle.next, 0);
}
@@ -2980,7 +3088,7 @@ fn fenced_usage_bootstrap_retains_partial_cycle_progress() {
},
PersistedUsageFloorStartup::BootstrapPending,
),
(true, false)
(true, ScannerCycleResetPolicy::None)
);
assert_eq!(cycle.next, 12);
@@ -2993,7 +3101,7 @@ fn fenced_usage_bootstrap_retains_partial_cycle_progress() {
},
PersistedUsageFloorStartup::Authoritative,
),
(false, false)
(false, ScannerCycleResetPolicy::None)
);
assert_eq!(cycle.next, 12);
}
@@ -3024,7 +3132,7 @@ async fn missing_usage_floor_rebuilds_persisted_cycle_before_leadership_claim()
.await
.expect("stably missing usage floor should admit a bootstrap marker");
assert_eq!(startup, PersistedUsageFloorStartup::Missing);
let (allow_bootstrap_pending, reset_bootstrap_cycle_on_conflict) =
let (allow_bootstrap_pending, cycle_reset_policy) =
prepare_cycle_for_usage_floor_bootstrap(&mut cycle_info, usage_floor, startup);
apply_persisted_usage_floor(&mut cycle_info, &mut persisted_epoch, usage_floor);
initialize_usage_baseline_bootstrap(store.clone())
@@ -3039,7 +3147,7 @@ async fn missing_usage_floor_rebuilds_persisted_cycle_before_leadership_claim()
&mut cycle_revision,
&mut persisted_epoch,
allow_bootstrap_pending,
reset_bootstrap_cycle_on_conflict,
cycle_reset_policy,
)
.await
);
@@ -3402,7 +3510,18 @@ async fn test_leadership_claim_preserves_usage_epoch_floor_across_old_epoch_conf
);
let mut persisted_epoch = 8;
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
assert!(
claim_scanner_leadership(
&ctx,
store.clone(),
&mut cycle,
&mut revision,
&mut persisted_epoch,
false,
ScannerCycleResetPolicy::None,
)
.await
);
let state = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH)
.await
@@ -3440,7 +3559,18 @@ async fn unfenced_usage_bootstrap_discards_old_epoch_conflict_progress() {
);
let mut persisted_epoch = 1;
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, true, true).await);
assert!(
claim_scanner_leadership(
&ctx,
store.clone(),
&mut cycle,
&mut revision,
&mut persisted_epoch,
true,
ScannerCycleResetPolicy::ResetAll,
)
.await
);
let state = read_config(store, &DATA_USAGE_BLOOM_NAME_PATH)
.await
@@ -3450,6 +3580,84 @@ async fn unfenced_usage_bootstrap_discards_old_epoch_conflict_progress() {
assert_eq!(claimed_epoch, 2);
}
#[tokio::test]
async fn recovered_usage_bootstrap_claim_conflicts_preserve_the_highest_cycle_number() {
for winner_next in [42_u64, 20_000] {
let store = Arc::new(MemoryConfigStore::default());
let ctx = CancellationToken::new();
let mut revision = DataUsageCacheRevision::Missing;
let mut cycle = CurrentCycle {
next: 12,
..Default::default()
};
assert!(persist_scanner_cycle_state(&ctx, store.clone(), &mut cycle, &mut revision, 1).await);
seed_usage_snapshot_for_leadership_claim(&store).await;
cycle = CurrentCycle {
current: 17_117,
next: 17_118,
cycle_completed: vec![Utc::now()],
started: Utc::now(),
};
let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
let winner = CurrentCycle {
current: winner_next.saturating_sub(1),
next: winner_next,
cycle_completed: vec![Utc::now()],
started: Utc::now(),
};
store
.interleaving_puts
.lock()
.await
.insert(key, (2, encode_scanner_cycle_state(&winner, 7).expect("conflict winner should encode")));
let mut persisted_epoch = 7;
assert!(
claim_scanner_leadership(
&ctx,
store.clone(),
&mut cycle,
&mut revision,
&mut persisted_epoch,
true,
ScannerCycleResetPolicy::ResetCoveragePreservingNext,
)
.await
);
let persisted = read_config(store, DATA_USAGE_BLOOM_NAME_PATH.as_str())
.await
.expect("recovered leadership claim should remain durable");
let (persisted_cycle, claimed_epoch) =
decode_scanner_cycle_state(&persisted).expect("recovered leadership claim should decode");
assert_eq!(persisted_cycle.next, 17_118_u64.max(winner_next));
assert_eq!(persisted_cycle.current, 0);
assert!(persisted_cycle.cycle_completed.is_empty());
assert_eq!(claimed_epoch, 8);
}
}
#[test]
fn recovered_usage_cache_reset_keeps_cycle_and_leader_regression_guards() {
let source = DataUsageCacheSource::new(0, 0);
let digest = DataUsageScanPlanDigest([9; 32]);
let mut newer_cycle = DataUsageCache::default();
newer_cycle.info.next_cycle = 17_119;
assert_eq!(
newer_cycle.prepare_for_scan(DATA_USAGE_ROOT, 17_118, 8, source, digest, true),
DataUsageCachePrepareOutcome::RejectedNewerCycle
);
let mut newer_leader = DataUsageCache::default();
newer_leader.info.next_cycle = 17_118;
newer_leader.info.leader_epoch = 9;
assert_eq!(
newer_leader.prepare_for_scan(DATA_USAGE_ROOT, 17_118, 8, source, digest, true),
DataUsageCachePrepareOutcome::RejectedNewerLeader
);
}
#[tokio::test]
async fn test_leadership_claim_rejects_terminal_epoch() {
let store = Arc::new(MemoryConfigStore::default());
@@ -3461,7 +3669,18 @@ async fn test_leadership_claim_rejects_terminal_epoch() {
};
let mut persisted_epoch = u64::MAX - 1;
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
assert!(
!claim_scanner_leadership(
&ctx,
store.clone(),
&mut cycle,
&mut revision,
&mut persisted_epoch,
false,
ScannerCycleResetPolicy::None,
)
.await
);
assert_eq!(persisted_epoch, u64::MAX - 1);
assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
}
@@ -3477,7 +3696,18 @@ async fn scanner_defers_leadership_when_usage_snapshots_are_stably_absent() {
};
let mut persisted_epoch = 0;
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
assert!(
!claim_scanner_leadership(
&ctx,
store.clone(),
&mut cycle,
&mut revision,
&mut persisted_epoch,
false,
ScannerCycleResetPolicy::None,
)
.await
);
assert!(read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
assert!(read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str()).await.is_err());
}
@@ -3493,9 +3723,31 @@ async fn usage_bootstrap_pending_unblocks_first_leadership_claim() {
let mut revision = DataUsageCacheRevision::Missing;
let mut cycle = CurrentCycle::default();
let mut persisted_epoch = 0;
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false,).await);
assert!(
!claim_scanner_leadership(
&ctx,
store.clone(),
&mut cycle,
&mut revision,
&mut persisted_epoch,
false,
ScannerCycleResetPolicy::None,
)
.await
);
assert!(read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, true, true).await);
assert!(
claim_scanner_leadership(
&ctx,
store.clone(),
&mut cycle,
&mut revision,
&mut persisted_epoch,
true,
ScannerCycleResetPolicy::ResetAll,
)
.await
);
let usage = read_config(store, DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
@@ -3583,7 +3835,18 @@ async fn leadership_claim_defers_on_corrupt_usage_baseline_without_bloom_write()
};
let mut persisted_epoch = 0;
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
assert!(
!claim_scanner_leadership(
&ctx,
store.clone(),
&mut cycle,
&mut revision,
&mut persisted_epoch,
false,
ScannerCycleResetPolicy::None,
)
.await
);
assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
}
@@ -3603,7 +3866,18 @@ async fn leadership_claim_defers_on_unidentified_usage_baseline_without_bloom_wr
};
let mut persisted_epoch = 0;
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
assert!(
!claim_scanner_leadership(
&ctx,
store.clone(),
&mut cycle,
&mut revision,
&mut persisted_epoch,
false,
ScannerCycleResetPolicy::None,
)
.await
);
assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
}
@@ -3625,7 +3899,18 @@ async fn test_leadership_claim_confirms_commit_after_returned_error() {
let mut persisted_epoch = 0;
seed_usage_snapshot_for_leadership_claim(&store).await;
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
assert!(
claim_scanner_leadership(
&ctx,
store.clone(),
&mut cycle,
&mut revision,
&mut persisted_epoch,
false,
ScannerCycleResetPolicy::None,
)
.await
);
let state = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH)
.await
@@ -3681,7 +3966,18 @@ async fn test_leadership_claim_usage_fence_rejects_old_inflight_writer() {
..Default::default()
};
let mut persisted_epoch = 4;
assert!(claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch, false, false).await);
assert!(
claim_scanner_leadership(
&ctx,
store.clone(),
&mut cycle,
&mut revision,
&mut persisted_epoch,
false,
ScannerCycleResetPolicy::None,
)
.await
);
let (fenced_data, fenced_revision) = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
@@ -3747,7 +4043,7 @@ async fn cycle_budget_lease_takeover_rejects_old_generation() {
&mut replacement_revision,
&mut replacement_epoch,
false,
false,
ScannerCycleResetPolicy::None,
)
.await
);