Merge branch 'main' into test/1325-fault-proxy-wiring

crates/ecstore/src/bucket/utils.rs

@@ -73,6 +73,7 @@ pub fn check_valid_bucket_name_strict(bucket_name: &str) -> Result<()> {
     check_bucket_name_common(bucket_name, true)
 }
 
+// RUSTFS_COMPAT_TODO(s3gate-metadata-xml): the s3s codec reads persisted XML during migration. Remove after every supported writer uses the gateway codec and every retained metadata object and backup archive is verified or rewritten.
 pub fn deserialize<T>(input: &[u8]) -> xml::DeResult<T>
 where
     T: for<'xml> xml::Deserialize<'xml>,

crates/ecstore/src/store/mod.rs

@@ -1205,6 +1205,23 @@ mod tests {
     use serial_test::serial;
     use tempfile::TempDir;
 
+    #[test]
+    fn g_d2_008_default_versioning_config_keeps_persisted_bytes() {
+        let bytes = crate::bucket::utils::serialize::<VersioningConfiguration>(&ENABLED_VERSIONING_CONFIG)
+            .expect("the default Versioning configuration must serialize");
+        assert_eq!(bytes, b"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>");
+    }
+
+    #[test]
+    fn g_d2_009_default_object_lock_config_keeps_persisted_bytes() {
+        let bytes = crate::bucket::utils::serialize::<ObjectLockConfiguration>(&ENABLED_OBJECT_LOCK_CONFIG)
+            .expect("the default Object Lock configuration must serialize");
+        assert_eq!(
+            bytes,
+            b"<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled></ObjectLockConfiguration>"
+        );
+    }
+
     #[tokio::test]
     async fn test_get_disk_infos() {
         let disks = vec![None, None]; // Empty disks for testing

crates/scanner/src/scanner.rs

@@ -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

@@ -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

@@ -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

@@ -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
     );

docs/architecture/compat-cleanup-register.md

@@ -13,6 +13,7 @@ for later deletion.
 ## Open Items
 
 - `backlog-2102` rc.2/rc.3 empty scanner usage floor recovery: old DeleteBucket cleanup could synthesize an empty incomplete v2 usage primary/backup before leadership added an epoch, while newer scanners require a durable authoritative baseline identity. New scanners recognize only that exact serialized empty-fence shape, preserve its epoch through a CAS-protected recovery marker, and rebuild namespace coverage without treating zero usage as authoritative. Remove this recovery path and marker after rc.2 and rc.3 are no longer supported direct-upgrade sources.
+- `s3gate-metadata-xml` persisted bucket XML migration: mixed-version site-replication peers, retained `.metadata.bin` objects, and backup archives can all carry XML written by the s3s codec, so the gateway migration must keep the legacy codec available until every stored form has crossed a verified rewrite boundary. Remove the legacy s3s parser and serializer only after the minimum supported direct-upgrade release reads and writes every persisted XML configuration family through the gateway codec, the four-way D1-D5 gate has remained clean for one full support window, every supported mixed-version site-replication topology has completed its writer upgrade, and migration tooling has verified or rewritten every retained bucket metadata object and restorable backup archive.
 - `rustfs-6339` legacy bucket policy ID casing: earlier RustFS releases persisted the top-level policy identifier as "ID", while current writes use the S3-compatible "Id" spelling. Readers accept both spellings so retained bucket metadata remains usable after upgrade. Remove the legacy alias after migration tooling has rewritten every retained bucket policy using "ID".
 - `table-publication-fence-v1` table publication fencing: nodes that predate table and table-bucket publication fences can mutate live files while a new node is publishing a catalog pointer. New nodes retain exact object guards until the operator confirms that every serving node uses the new fences. Fleet confirmation also requires non-overlapping active warehouse prefixes and lifecycle workers that exclude table buckets. Remove the exact live-file fallback and the fleet-confirmation gate after the minimum supported RustFS release acquires table fences for registered-table mutations and table-bucket fences for unresolved-prefix mutations.
 - `table-catalog-strong-snapshot-v1` durable strong catalog snapshot compatibility: version 1 writes continue during mixed-version rollout until operators confirm that every serving node reads version 2, and version 1 table/view identifier collisions remain available only for cleanup. Remove version 1 writes and collision cleanup after the minimum supported RustFS release reads version 2 and every retained durable strong snapshot is collision-free and has been upgraded to version 2.