docs(scanner): clarify unknown tier stats

* fix(scanner): fence timed out scan cycles

* fix(scanner): cancel scan workers with cycle scope

* fix(scanner): reject persisted timer overflow

* fix(scanner): reject terminal leadership epochs

* fix(scanner): reject trailing cycle state bytes

Cargo.lock

@@ -9587,7 +9587,6 @@ dependencies = [
  "serde",
  "serde_json",
  "serial_test",
- "sha2 0.11.0",
  "temp-env",
  "tempfile",
  "thiserror 2.0.20",

crates/common/src/metrics.rs

@@ -901,6 +901,10 @@ pub struct Metrics {
     scanner_cycle_max_duration_millis: AtomicU64,
     scanner_cycle_max_objects: AtomicU64,
     scanner_cycle_max_directories: AtomicU64,
+    scanner_cycle_timeout_total: AtomicU64,
+    scanner_cycle_recovery_required_total: AtomicU64,
+    scanner_cycle_last_progress_age_seconds: AtomicU64,
+    scanner_leader_lease_without_progress: AtomicBool,
     scanner_bitrot_cycle_enabled: AtomicBool,
     scanner_bitrot_cycle_millis: AtomicU64,
     scanner_checkpoint: Mutex<Option<ScannerCheckpointReport>>,
@@ -1370,6 +1374,14 @@ pub struct ScannerMetricsReport {
     #[serde(default)]
     pub cycle_max_directories: u64,
     #[serde(default)]
+    pub cycle_timeout_total: u64,
+    #[serde(default)]
+    pub cycle_recovery_required_total: u64,
+    #[serde(default)]
+    pub cycle_last_progress_age: u64,
+    #[serde(default)]
+    pub leader_lease_without_progress: bool,
+    #[serde(default)]
     pub bitrot_cycle_enabled: bool,
     #[serde(default)]
     pub bitrot_cycle_seconds: f64,
@@ -1430,6 +1442,9 @@ const OTEL_SCANNER_BUCKETS_SCANNED: &str = "rustfs_scanner_buckets_scanned_total
 const OTEL_SCANNER_CYCLES: &str = "rustfs_scanner_cycles_total";
 const OTEL_SCANNER_CYCLE_DURATION_SECONDS: &str = "rustfs_scanner_cycle_duration_seconds";
 const OTEL_SCANNER_BUCKET_DRIVE_DURATION_SECONDS: &str = "rustfs_scanner_bucket_drive_duration_seconds";
+const OTEL_SCANNER_CYCLE_TIMEOUT_TOTAL: &str = "rustfs_scanner_cycle_timeout_total";
+const OTEL_SCANNER_CYCLE_LAST_PROGRESS_AGE: &str = "rustfs_scanner_cycle_last_progress_age";
+const OTEL_SCANNER_LEADER_LEASE_WITHOUT_PROGRESS: &str = "rustfs_scanner_leader_lease_without_progress";
 
 fn scan_cycle_result_label(result: u8) -> &'static str {
     match result {
@@ -1913,6 +1928,10 @@ impl Metrics {
             scanner_cycle_max_duration_millis: AtomicU64::new(0),
             scanner_cycle_max_objects: AtomicU64::new(0),
             scanner_cycle_max_directories: AtomicU64::new(0),
+            scanner_cycle_timeout_total: AtomicU64::new(0),
+            scanner_cycle_recovery_required_total: AtomicU64::new(0),
+            scanner_cycle_last_progress_age_seconds: AtomicU64::new(0),
+            scanner_leader_lease_without_progress: AtomicBool::new(false),
             scanner_bitrot_cycle_enabled: AtomicBool::new(false),
             scanner_bitrot_cycle_millis: AtomicU64::new(0),
             scanner_checkpoint: Mutex::new(None),
@@ -2412,12 +2431,29 @@ impl Metrics {
             .store(cycle_max_objects.unwrap_or_default(), Ordering::Relaxed);
         self.scanner_cycle_max_directories
             .store(cycle_max_directories.unwrap_or_default(), Ordering::Relaxed);
+        self.scanner_leader_lease_without_progress.store(false, Ordering::Relaxed);
+        self.scanner_cycle_last_progress_age_seconds.store(0, Ordering::Relaxed);
+        metrics::gauge!(OTEL_SCANNER_LEADER_LEASE_WITHOUT_PROGRESS).set(0.0);
+        metrics::gauge!(OTEL_SCANNER_CYCLE_LAST_PROGRESS_AGE).set(0.0);
         self.scanner_bitrot_cycle_enabled
             .store(bitrot_cycle.is_some(), Ordering::Relaxed);
         self.scanner_bitrot_cycle_millis
             .store(bitrot_cycle.map(duration_millis_saturated).unwrap_or_default(), Ordering::Relaxed);
     }
 
+    pub fn record_scanner_cycle_timeout(&self, recovery_required: bool, progress_age: Duration) {
+        self.scanner_cycle_timeout_total.fetch_add(1, Ordering::Relaxed);
+        if recovery_required {
+            self.scanner_cycle_recovery_required_total.fetch_add(1, Ordering::Relaxed);
+        }
+        self.scanner_cycle_last_progress_age_seconds
+            .store(progress_age.as_secs(), Ordering::Relaxed);
+        self.scanner_leader_lease_without_progress.store(true, Ordering::Relaxed);
+        metrics::counter!(OTEL_SCANNER_CYCLE_TIMEOUT_TOTAL).increment(1);
+        metrics::gauge!(OTEL_SCANNER_CYCLE_LAST_PROGRESS_AGE).set(progress_age.as_secs_f64());
+        metrics::gauge!(OTEL_SCANNER_LEADER_LEASE_WITHOUT_PROGRESS).set(1.0);
+    }
+
     pub fn record_scanner_set_scan_state(&self, concurrency_limit: Option<usize>, queued: Option<usize>, active: Option<usize>) {
         if let Some(concurrency_limit) = concurrency_limit {
             self.scanner_set_scan_concurrency_limit
@@ -3265,6 +3301,10 @@ impl Metrics {
         m.cycle_max_duration_seconds = self.scanner_cycle_max_duration_millis.load(Ordering::Relaxed) as f64 / 1000.0;
         m.cycle_max_objects = self.scanner_cycle_max_objects.load(Ordering::Relaxed);
         m.cycle_max_directories = self.scanner_cycle_max_directories.load(Ordering::Relaxed);
+        m.cycle_timeout_total = self.scanner_cycle_timeout_total.load(Ordering::Relaxed);
+        m.cycle_recovery_required_total = self.scanner_cycle_recovery_required_total.load(Ordering::Relaxed);
+        m.cycle_last_progress_age = self.scanner_cycle_last_progress_age_seconds.load(Ordering::Relaxed);
+        m.leader_lease_without_progress = self.scanner_leader_lease_without_progress.load(Ordering::Relaxed);
         m.bitrot_cycle_enabled = self.scanner_bitrot_cycle_enabled.load(Ordering::Relaxed);
         m.bitrot_cycle_seconds = self.scanner_bitrot_cycle_millis.load(Ordering::Relaxed) as f64 / 1000.0;
         m.scan_checkpoint = match self.scanner_checkpoint.lock() {
@@ -4926,4 +4966,20 @@ mod tests {
         assert!(!report.bitrot_cycle_enabled);
         assert_eq!(report.bitrot_cycle_seconds, 0.0);
     }
+
+    #[tokio::test]
+    async fn scanner_cycle_timeout_metrics_reset_for_a_new_cycle() {
+        let metrics = Metrics::new();
+        metrics.record_scanner_cycle_timeout(true, Duration::from_secs(17));
+        let timed_out = metrics.report().await;
+        assert_eq!(timed_out.cycle_timeout_total, 1);
+        assert_eq!(timed_out.cycle_last_progress_age, 17);
+        assert!(timed_out.leader_lease_without_progress);
+
+        metrics.record_scanner_cycle_config(Duration::from_secs(60), None, Some(Duration::from_secs(1)), None, None);
+        let current = metrics.report().await;
+        assert_eq!(current.cycle_timeout_total, 1);
+        assert_eq!(current.cycle_last_progress_age, 0);
+        assert!(!current.leader_lease_without_progress);
+    }
 }

crates/config/README.md

@@ -84,6 +84,12 @@ Current guidance:
 - `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` (canonical)
 - `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` (canonical)
 
+Scanner cycle budget controls:
+
+- When `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` is unset, the finite default is 1800 seconds (30 minutes), matching the scanner benchmark guidance.
+- An explicit `0` preserves the compatibility behavior of an unbounded runtime budget. Object and directory budgets likewise remain unbounded when explicitly set to `0`.
+- A timed-out cycle cancels cooperative scanner work, then fences its leader epoch before releasing the lease. An uncooperative I/O operation is dropped after the bounded shutdown window; its cursor is not claimed to be durable and the scanner reports `recovery-required` when the worker cannot stop cooperatively, the cycle state was not confirmed durable, or epoch fencing cannot be persisted.
+
 ## Mmap read environment aliases
 
 - `RUSTFS_OBJECT_MMAP_READ_ENABLE` (canonical)

crates/config/src/constants/scanner.rs

@@ -143,9 +143,12 @@ pub const ENV_SCANNER_MAX_WAIT_SECS: &str = "RUSTFS_SCANNER_MAX_WAIT_SECS";
 /// Default scanner speed preset.
 pub const DEFAULT_SCANNER_SPEED: &str = "default";
 
-/// Default scanner cycle runtime budget.
-/// `0` keeps the existing unbounded per-cycle behavior.
-pub const DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS: u64 = 0;
+/// Default scanner cycle runtime budget when no override is configured.
+///
+/// An explicit `0` remains the compatibility escape hatch for an unbounded
+/// cycle. Keeping the unset default finite prevents a stalled scanner I/O
+/// operation from holding the leader lease forever.
+pub const DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS: u64 = 30 * 60;
 
 /// Default scanner per-cycle object budget.
 /// `0` keeps the existing unbounded per-cycle behavior.

crates/data-usage/src/data_usage.rs

@@ -48,6 +48,11 @@ pub const DATA_USAGE_OBSERVED_OBJECT_NAME: &str = ".usage.observed.json";
 // RUSTFS_COMPAT_TODO(scanner-usage-v2): keep .usage.json readable and removable during rolling upgrades from pre-v2 scanners. Remove after supported direct-upgrade sources all write .usage.v2.json.
 pub const LEGACY_DATA_USAGE_OBJECT_NAME: &str = ".usage.json";
 
+/// Fixed bucket for objects whose storage class is not in the scanner's
+/// cycle-local tier registry.  Keeping this key fixed prevents untrusted or
+/// stale tier names from growing persisted per-tier maps without bound.
+pub const UNKNOWN_TIER: &str = "UNKNOWN_TIER";
+
 /// Returns true when `existing_last_update` is ahead of `now` by more than
 /// [`USAGE_LAST_UPDATE_FUTURE_TOLERANCE`], i.e. the persisted timestamp cannot be
 /// trusted for staleness comparisons and a fresh snapshot save must be allowed.
@@ -78,12 +83,146 @@ impl TierStats {
             && self.num_objects.checked_add(u.num_objects).is_some()
     }
 
+    /// Add tier counters without allowing a counter to wrap.
+    pub fn checked_add(&self, u: &TierStats) -> Option<TierStats> {
+        Some(TierStats {
+            total_size: self.total_size.checked_add(u.total_size)?,
+            num_versions: self.num_versions.checked_add(u.num_versions)?,
+            num_objects: self.num_objects.checked_add(u.num_objects)?,
+        })
+    }
+
     /// True when this tier contributed nothing, i.e. merging it is a no-op.
     pub fn is_empty(&self) -> bool {
         self.total_size == 0 && self.num_versions == 0 && self.num_objects == 0
     }
 }
 
+/// Bounded diagnostics for objects whose tier is absent from the cycle
+/// registry. Counters are authoritative; diagnostics are only a small,
+/// redacted reconciliation aid and may be dropped at the configured caps.
+pub const UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP: usize = 64;
+pub const UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP: usize = 4096;
+pub const UNKNOWN_TIER_DIAGNOSTIC_TTL: Duration = Duration::from_secs(60 * 60);
+const UNKNOWN_TIER_DIAGNOSTIC_KEY_BYTES: usize = 256;
+
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct UnknownTierStats {
+    /// Logical bytes retained in the scanner's normal usage total.
+    pub unknown_bytes: u64,
+    /// Physical bytes recorded in the per-tier accounting dimension.
+    ///
+    /// Older writers only had `unknown_bytes`; decoding those snapshots keeps
+    /// this field at zero and the scanner fills it for new observations.
+    #[serde(default)]
+    pub unknown_physical_bytes: u64,
+    pub unknown_objects: u64,
+    pub unknown_versions: u64,
+    pub diagnostics_dropped: u64,
+    #[serde(default)]
+    pub diagnostics: Vec<String>,
+    #[serde(default)]
+    pub diagnostics_at: Option<SystemTime>,
+}
+
+impl UnknownTierStats {
+    /// Record one observation where the logical and physical dimensions are
+    /// the same. Kept as a small compatibility helper for callers that only
+    /// have one size value.
+    pub fn record(&mut self, tier: &str, bytes: u64, versions: u64, objects: u64) {
+        self.record_dimensions(tier, bytes, bytes, versions, objects);
+    }
+
+    /// Record one observation without conflating logical usage with physical
+    /// tier bytes. Both counters are saturating so malformed metadata cannot
+    /// wrap an aggregate.
+    pub fn record_dimensions(&mut self, tier: &str, logical_bytes: u64, physical_bytes: u64, versions: u64, objects: u64) {
+        self.unknown_bytes = self.unknown_bytes.saturating_add(logical_bytes);
+        self.unknown_physical_bytes = self.unknown_physical_bytes.saturating_add(physical_bytes);
+        self.unknown_objects = self.unknown_objects.saturating_add(objects);
+        self.unknown_versions = self.unknown_versions.saturating_add(versions);
+
+        let digest = {
+            let mut hasher = DefaultHasher::new();
+            // Bound hashing work for hostile metadata while retaining enough
+            // length/prefix entropy to reconcile repeated observations.
+            hasher.write_usize(tier.len());
+            hasher.write(&tier.as_bytes()[..tier.len().min(UNKNOWN_TIER_DIAGNOSTIC_KEY_BYTES)]);
+            hasher.write_u8(u8::from(tier.bytes().any(|byte| byte.is_ascii_control())));
+            format!("tier-hash:{:016x}", hasher.finish())
+        };
+        let now = SystemTime::now();
+        if self
+            .diagnostics_at
+            .is_some_and(|at| now.duration_since(at).unwrap_or_default() > UNKNOWN_TIER_DIAGNOSTIC_TTL)
+        {
+            self.diagnostics.clear();
+        }
+        self.diagnostics_at = Some(now);
+        if self.diagnostics.iter().any(|entry| entry == &digest) {
+            return;
+        }
+        let current_bytes: usize = self.diagnostics.iter().map(String::len).sum();
+        if self.diagnostics.len() >= UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP
+            || current_bytes.saturating_add(digest.len()) > UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP
+        {
+            self.diagnostics_dropped = self.diagnostics_dropped.saturating_add(1);
+            return;
+        }
+        self.diagnostics.push(digest);
+    }
+
+    pub fn merge(&mut self, other: &Self) {
+        self.unknown_bytes = self.unknown_bytes.saturating_add(other.unknown_bytes);
+        self.unknown_physical_bytes = self.unknown_physical_bytes.saturating_add(other.unknown_physical_bytes);
+        self.unknown_objects = self.unknown_objects.saturating_add(other.unknown_objects);
+        self.unknown_versions = self.unknown_versions.saturating_add(other.unknown_versions);
+        self.diagnostics_dropped = self.diagnostics_dropped.saturating_add(other.diagnostics_dropped);
+        let now = SystemTime::now();
+        if self
+            .diagnostics_at
+            .zip(other.diagnostics_at)
+            .is_some_and(|(left, right)| now.duration_since(left.max(right)).unwrap_or_default() > UNKNOWN_TIER_DIAGNOSTIC_TTL)
+        {
+            self.diagnostics.clear();
+        }
+        self.diagnostics_at = Some(now);
+        for diagnostic in &other.diagnostics {
+            if self.diagnostics.iter().any(|entry| entry == diagnostic) {
+                continue;
+            }
+            let current_bytes: usize = self.diagnostics.iter().map(String::len).sum();
+            if self.diagnostics.len() >= UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP
+                || current_bytes.saturating_add(diagnostic.len()) > UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP
+            {
+                self.diagnostics_dropped = self.diagnostics_dropped.saturating_add(1);
+                continue;
+            }
+            self.diagnostics.push(diagnostic.clone());
+        }
+    }
+
+    pub fn fits_add(&self, other: &Self) -> bool {
+        self.unknown_bytes.checked_add(other.unknown_bytes).is_some()
+            && self
+                .unknown_physical_bytes
+                .checked_add(other.unknown_physical_bytes)
+                .is_some()
+            && self.unknown_objects.checked_add(other.unknown_objects).is_some()
+            && self.unknown_versions.checked_add(other.unknown_versions).is_some()
+            && self.diagnostics_dropped.checked_add(other.diagnostics_dropped).is_some()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.unknown_bytes == 0
+            && self.unknown_physical_bytes == 0
+            && self.unknown_objects == 0
+            && self.unknown_versions == 0
+            && self.diagnostics_dropped == 0
+            && self.diagnostics.is_empty()
+    }
+}
+
 #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
 pub struct AllTierStats {
     pub tiers: HashMap<String, TierStats>,
@@ -124,6 +263,31 @@ impl AllTierStats {
             .iter()
             .all(|(tier, right)| self.tiers.get(tier).is_none_or(|left| left.fits_add(right)))
     }
+
+    /// Fold keys from an older cache that are no longer present in the
+    /// current registry into the fixed unknown bucket. Built-in storage
+    /// classes remain known even when no remote tier is configured.
+    pub fn fold_unknown_tiers<'a, I>(&mut self, known_tiers: I)
+    where
+        I: IntoIterator<Item = &'a str>,
+    {
+        let known: HashSet<&str> = known_tiers.into_iter().collect();
+        let mut unknown = self.tiers.remove(UNKNOWN_TIER).unwrap_or_default();
+        let retired: Vec<String> = self
+            .tiers
+            .keys()
+            .filter(|tier| tier.as_str() != "STANDARD" && tier.as_str() != "REDUCED_REDUNDANCY" && !known.contains(tier.as_str()))
+            .cloned()
+            .collect();
+        for tier in retired {
+            if let Some(stats) = self.tiers.remove(&tier) {
+                unknown = unknown.add(&stats);
+            }
+        }
+        if !unknown.is_empty() {
+            self.tiers.insert(UNKNOWN_TIER.to_string(), unknown);
+        }
+    }
 }
 
 /// Bucket target usage info provides replication statistics
@@ -211,6 +375,10 @@ pub struct DataUsageInfo {
     /// tier exists, so an absent value means "not accounted", never "zero".
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub tier_stats: Option<AllTierStats>,
+    /// Bounded diagnostics and separate logical/physical counters for objects
+    /// classified into [`UNKNOWN_TIER`].
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub unknown_tier_stats: Option<UnknownTierStats>,
 
     /// Total number of buckets in this cluster
     pub buckets_count: u64,
@@ -336,6 +504,8 @@ pub struct SizeSummary {
     pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
     /// Per-tier accounting, keyed by storage class or remote tier name
     pub tier_stats: HashMap<String, TierStats>,
+    /// Counters and bounded diagnostics for unknown tiers in this summary.
+    pub unknown_tier_stats: UnknownTierStats,
 }
 
 /// Replication target size summary
@@ -681,6 +851,9 @@ pub struct DataUsageEntry {
     /// observed tier-classified objects.
     #[serde(default)]
     pub all_tier_stats: Option<AllTierStats>,
+    /// Bounded unknown-tier reconciliation state for this cache entry.
+    #[serde(default)]
+    pub unknown_tier_stats: Option<UnknownTierStats>,
 }
 
 impl Serialize for DataUsageEntry {
@@ -691,7 +864,7 @@ impl Serialize for DataUsageEntry {
         // Keep entries map-encoded so older readers can ignore fields appended
         // by newer scanner versions during rolling upgrades. The derived
         // (array) encoding made any appended field a decode error for them.
-        let mut state = serializer.serialize_map(Some(11))?;
+        let mut state = serializer.serialize_map(Some(12))?;
         state.serialize_entry("children", &self.children)?;
         state.serialize_entry("size", &self.size)?;
         state.serialize_entry("objects", &self.objects)?;
@@ -703,6 +876,7 @@ impl Serialize for DataUsageEntry {
         state.serialize_entry("compacted", &self.compacted)?;
         state.serialize_entry("failed_objects", &self.failed_objects)?;
         state.serialize_entry("all_tier_stats", &self.all_tier_stats)?;
+        state.serialize_entry("unknown_tier_stats", &self.unknown_tier_stats)?;
         state.end()
     }
 }
@@ -744,6 +918,11 @@ impl DataUsageEntry {
         if let Some(o_tiers) = other.all_tier_stats.as_ref().filter(|tiers| !tiers.is_empty()) {
             self.all_tier_stats.get_or_insert_with(AllTierStats::new).merge(o_tiers);
         }
+        if let Some(other_unknown) = other.unknown_tier_stats.as_ref() {
+            self.unknown_tier_stats
+                .get_or_insert_with(UnknownTierStats::default)
+                .merge(other_unknown);
+        }
 
         self.obj_sizes.merge_from(&other.obj_sizes);
         self.obj_versions.merge_from(&other.obj_versions);
@@ -757,6 +936,15 @@ impl DataUsageEntry {
         self.all_tier_stats.get_or_insert_with(AllTierStats::new).add_sizes(tiers);
     }
 
+    pub fn add_unknown_tier_stats(&mut self, stats: &UnknownTierStats) {
+        if stats.is_empty() {
+            return;
+        }
+        self.unknown_tier_stats
+            .get_or_insert_with(UnknownTierStats::default)
+            .merge(stats);
+    }
+
     pub fn checked_merge(&mut self, other: &DataUsageEntry) -> bool {
         let scalar_counts_fit = self.objects.checked_add(other.objects).is_some()
             && self.versions.checked_add(other.versions).is_some()
@@ -820,8 +1008,12 @@ impl DataUsageEntry {
             (_, None) | (None, Some(_)) => true,
             (Some(left), Some(right)) => left.fits_merge(right),
         };
+        let unknown_tier_stats_fit = match (&self.unknown_tier_stats, &other.unknown_tier_stats) {
+            (_, None) | (None, Some(_)) => true,
+            (Some(left), Some(right)) => left.fits_add(right),
+        };
 
-        if !scalar_counts_fit || !histograms_fit || !replication_fits || !tier_stats_fit {
+        if !scalar_counts_fit || !histograms_fit || !replication_fits || !tier_stats_fit || !unknown_tier_stats_fit {
             return false;
         }
         self.merge(other);
@@ -1339,6 +1531,7 @@ impl DataUsageCache {
             delete_markers_total_count: flat.delete_markers as u64,
             objects_total_size: flat.size as u64,
             tier_stats: flat.all_tier_stats.filter(|tiers| !tiers.is_empty()),
+            unknown_tier_stats: flat.unknown_tier_stats,
             buckets_count: u64::try_from(buckets.len()).unwrap_or(u64::MAX),
             buckets_usage,
             usage_snapshot_complete: self.info.snapshot_complete,
@@ -1766,6 +1959,16 @@ impl SizeSummary {
         self.replica_count = self.replica_count.saturating_add(other.replica_count);
         self.pending_count = self.pending_count.saturating_add(other.pending_count);
         self.failed_count = self.failed_count.saturating_add(other.failed_count);
+        self.unknown_tier_stats.merge(&other.unknown_tier_stats);
+
+        // A disk/bucket aggregate is assembled from many object summaries.
+        // Keep the per-tier dimension in lockstep with the scalar counters;
+        // dropping this map here would recreate the original silent-loss bug
+        // at the cross-disk merge boundary.
+        for (tier, stats) in &other.tier_stats {
+            let entry = self.tier_stats.entry(tier.clone()).or_default();
+            *entry = entry.add(stats);
+        }
 
         // Merge replication target stats
         for (target, stats) in &other.repl_target_stats {
@@ -1878,6 +2081,33 @@ mod tests {
         );
     }
 
+    #[test]
+    fn retired_tier_stats_fold_into_fixed_unknown_bucket() {
+        let mut stats = AllTierStats::default();
+        stats.tiers.insert(
+            "RETIRED".to_string(),
+            TierStats {
+                total_size: 9,
+                num_versions: 2,
+                num_objects: 1,
+            },
+        );
+        stats.tiers.insert(
+            "WARM".to_string(),
+            TierStats {
+                total_size: 4,
+                num_versions: 1,
+                num_objects: 1,
+            },
+        );
+
+        stats.fold_unknown_tiers(["WARM"]);
+
+        assert!(!stats.tiers.contains_key("RETIRED"));
+        assert_eq!(stats.tiers.get("WARM").map(|v| v.total_size), Some(4));
+        assert_eq!(stats.tiers.get(UNKNOWN_TIER).map(|v| v.total_size), Some(9));
+    }
+
     #[test]
     fn checked_merge_rejects_overflowing_tier_totals() {
         let mut left = tier_entry(

crates/ecstore/src/disk/local.rs

@@ -7949,15 +7949,10 @@ impl DiskAPI for LocalDisk {
             use std::io::Write as _;
 
             let file_path = self.io_get_object_path(volume, path)?;
+            let lock_path = file_path.with_extension("rustfs-cas.lock");
             let path = path.to_string();
             let sync_metadata = effective_durability(volume).syncs_commit_metadata();
             return Ok(tokio::task::spawn_blocking(move || {
-                // A persistent directory lock bounds metadata growth. Removing
-                // per-target lock files can split flock ownership across inodes.
-                let lock_path = file_path
-                    .parent()
-                    .ok_or_else(|| std::io::Error::new(ErrorKind::InvalidInput, "conditional file has no parent"))?
-                    .join(".rustfs-cas.lock");
                 let lock = std::fs::OpenOptions::new()
                     .create(true)
                     .truncate(false)
@@ -21846,10 +21841,7 @@ mod test {
         let marker_path = disk
             .get_object_path(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
             .expect("marker path should resolve");
-        let lock_path = marker_path
-            .parent()
-            .expect("marker path should have a parent")
-            .join(".rustfs-cas.lock");
+        let lock_path = marker_path.with_extension("rustfs-cas.lock");
         let lock = std::fs::OpenOptions::new()
             .create(true)
             .truncate(false)

crates/ecstore/src/services/metrics_realtime.rs

@@ -256,6 +256,10 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo
         cycle_max_duration_seconds: metrics.cycle_max_duration_seconds,
         cycle_max_objects: metrics.cycle_max_objects,
         cycle_max_directories: metrics.cycle_max_directories,
+        cycle_timeout_total: metrics.cycle_timeout_total,
+        cycle_recovery_required_total: metrics.cycle_recovery_required_total,
+        cycle_last_progress_age: metrics.cycle_last_progress_age,
+        leader_lease_without_progress: metrics.leader_lease_without_progress,
         bitrot_cycle_enabled: metrics.bitrot_cycle_enabled,
         bitrot_cycle_seconds: metrics.bitrot_cycle_seconds,
         scan_checkpoint: metrics.scan_checkpoint.map(|checkpoint| MadminScannerCheckpointReport {
@@ -611,6 +615,10 @@ mod test {
             current_started: chrono_to_jiff_timestamp(current_started),
             last_cycle_partial_source: "usage".to_string(),
             last_cycle_partial_source_code: 1,
+            cycle_timeout_total: 3,
+            cycle_recovery_required_total: 2,
+            cycle_last_progress_age: 17,
+            leader_lease_without_progress: true,
             partial_cycles_by_source: vec![rustfs_common::metrics::ScannerSourceCycleSnapshot {
                 source: "usage".to_string(),
                 cycles: 2,
@@ -622,6 +630,10 @@ mod test {
         assert_eq!(scanner.current_started, chrono_to_jiff_timestamp(current_started));
         assert_eq!(scanner.last_cycle_partial_source, "usage");
         assert_eq!(scanner.last_cycle_partial_source_code, 1);
+        assert_eq!(scanner.cycle_timeout_total, 3);
+        assert_eq!(scanner.cycle_recovery_required_total, 2);
+        assert_eq!(scanner.cycle_last_progress_age, 17);
+        assert!(scanner.leader_lease_without_progress);
         let usage = scanner
             .partial_cycles_by_source
             .iter()

crates/ecstore/src/store/mod.rs

@@ -33,7 +33,7 @@ use crate::bucket::utils::check_put_object_part_args;
 use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname};
 use crate::cluster::rpc::{RemoteClient, S3PeerSys};
 use crate::config::storageclass;
-use crate::core::pools::PoolMeta;
+use crate::core::pools::{DecommissionCanceler, PoolMeta};
 use crate::disk::endpoint::{Endpoint, EndpointType};
 use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions};
 use crate::error::{Error, Result};
@@ -176,7 +176,7 @@ pub struct ECStore {
     // pub local_disks: Vec<DiskStore>,
     pub pool_meta: RwLock<PoolMeta>,
     pub rebalance_meta: RwLock<Option<RebalanceMeta>>,
-    pub decommission_cancelers: RwLock<Vec<Option<CancellationToken>>>,
+    pub decommission_cancelers: RwLock<Vec<Option<DecommissionCanceler>>>,
     /// Serializes rebalance/decommission start transitions.
     ///
     /// Lock order: acquire `start_gate` before `pool_meta`, `rebalance_meta`,

crates/heal/Cargo.toml

@@ -91,7 +91,6 @@ metrics = { workspace = true }
 base64 = { workspace = true }
 bytes = { workspace = true }
 crc-fast = { workspace = true }
-sha2 = { workspace = true }
 
 [dev-dependencies]
 serde_json = { workspace = true, features = ["raw_value"] }

crates/heal/src/heal/erasure_healer.rs

@@ -373,11 +373,6 @@ impl ErasureSetHealer {
         set_disk_id: &str,
         buckets: &[String],
     ) -> Result<(ResumeManager, CheckpointManager)> {
-        if self.replacement_task_id.is_none() && CheckpointManager::is_blocked(&self.disk, task_id).await {
-            return Err(Error::TaskExecutionFailed {
-                message: format!("Resume task {task_id} has a blocked checkpoint"),
-            });
-        }
         // check if resume state exists
         let has_resume_state = if self.replacement_task_id.is_some() {
             ResumeManager::has_replacement_intent(&self.disk, task_id).await

crates/heal/src/heal/resume.rs

@@ -51,7 +51,6 @@ const RESUME_STATE_FILE: &str = "ahm_resume_state.json";
 const REPLACEMENT_INTENT_FILE: &str = "ahm_replacement_intent.json";
 const RESUME_PROGRESS_FILE: &str = "ahm_progress.json";
 pub(super) const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json";
-pub(super) const RESUME_CHECKPOINT_BLOCKED_FILE: &str = "ahm_checkpoint.blocked";
 const REPLACEMENT_COMPLETION_PROOF_FILE: &str = "ahm_replacement_completion_proof.json";
 const REPLACEMENT_RECOVERY_DIR: &str = "ahm-replacement";
 const REPLACEMENT_INTENT_SEAL_FILE: &str = "ahm_replacement_intent_seal";

crates/heal/src/heal/resume/checkpoint.rs

@@ -13,31 +13,26 @@
 // limitations under the License.
 
 use crate::{Error, Result};
-use base64::Engine as _;
 use serde::{Deserialize, Serialize};
-use sha2::{Digest, Sha256};
 use std::collections::HashSet;
 use std::path::Path;
 use std::sync::{Arc, Mutex};
 use std::time::{SystemTime, UNIX_EPOCH};
-use tokio::sync::{Mutex as AsyncMutex, RwLock};
+use tokio::sync::RwLock;
 use tracing::{debug, warn};
 
-use super::super::storage_api::owner::{EcstoreConditionalFileUpdate, EcstoreDiskAPI, EcstoreDiskBytes};
-use super::super::{BUCKET_META_PREFIX, DiskStore, HealDiskExt, RUSTFS_META_BUCKET};
+use super::super::{BUCKET_META_PREFIX, DiskStore, HealDiskExt as _, RUSTFS_META_BUCKET};
 use super::{
-    LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, PersistThrottle, RESUME_CHECKPOINT_BLOCKED_FILE, RESUME_CHECKPOINT_FILE,
-    delete_resume_file, path_to_str, validate_resume_task_id,
+    LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, PersistThrottle, RESUME_CHECKPOINT_FILE, delete_resume_file, path_to_str,
+    validate_resume_task_id,
 };
 
 const EVENT_HEAL_CHECKPOINT_STATE: &str = "heal_checkpoint_state";
-const RESUME_CHECKPOINT_DIGEST_FILE: &str = "ahm_checkpoint.sha256";
-const CHECKPOINT_PER_VERSION_SCHEMA: u32 = 5;
 
 /// Current on-disk schema version for `ResumeCheckpoint`. Same rationale as
 /// `CURRENT_RESUME_SCHEMA`: pre-per-version dedup identities are not comparable
 /// to the new `compose_key` identities, so a stale checkpoint is discarded.
-pub(super) const CURRENT_CHECKPOINT_SCHEMA: u32 = 6;
+pub(super) const CURRENT_CHECKPOINT_SCHEMA: u32 = 5;
 
 /// resume checkpoint
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -62,11 +57,6 @@ pub struct ResumeCheckpoint {
     pub failed_objects: HashSet<String>,
     /// skipped objects
     pub skipped_objects: HashSet<String>,
-    /// Integrity digest over the checkpoint with this field set to `None`.
-    /// Keeping it in the checkpoint makes the payload and its authentication
-    /// record one CAS generation instead of two independently-written files.
-    #[serde(default)]
-    pub integrity_digest: Option<String>,
 }
 
 impl ResumeCheckpoint {
@@ -80,7 +70,6 @@ impl ResumeCheckpoint {
             processed_objects: HashSet::new(),
             failed_objects: HashSet::new(),
             skipped_objects: HashSet::new(),
-            integrity_digest: None,
         }
     }
 
@@ -127,111 +116,17 @@ pub struct CheckpointManager {
     disk: DiskStore,
     checkpoint: Arc<RwLock<ResumeCheckpoint>>,
     throttle: Mutex<PersistThrottle>,
-    save_lock: AsyncMutex<()>,
-    last_saved: Mutex<Option<EcstoreDiskBytes>>,
 }
 
 impl CheckpointManager {
-    fn blocked_path(task_id: &str) -> std::path::PathBuf {
-        Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}"))
-    }
-
-    /// Return whether a checkpoint was permanently isolated after a malformed
-    /// or unsupported snapshot was observed.
-    pub(crate) async fn is_blocked(disk: &DiskStore, task_id: &str) -> bool {
-        if validate_resume_task_id(task_id).is_err() {
-            return false;
-        }
-        let blocked_path = Self::blocked_path(task_id);
-        let Ok(path) = path_to_str(&blocked_path) else {
-            return false;
-        };
-        match HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path).await {
-            Ok(_) => true,
-            Err(crate::heal::DiskError::FileNotFound) => false,
-            Err(_) => true,
-        }
-    }
-
-    /// Validate the checkpoint while enumerating resumable state. This reads
-    /// the checkpoint once and also isolates malformed or unsupported data.
-    pub(crate) async fn is_resumable(disk: &DiskStore, task_id: &str) -> Result<bool> {
-        validate_resume_task_id(task_id)?;
-        if Self::is_blocked(disk, task_id).await {
-            return Err(Error::InvalidCheckpoint(format!("Resume task {task_id} has a blocked checkpoint")));
-        }
-        let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"));
-        let Ok(path) = path_to_str(&file_path) else {
-            return Err(Error::InvalidCheckpoint("Resume checkpoint path is not valid UTF-8".to_string()));
-        };
-        match HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path).await {
-            Ok(bytes) if bytes.is_empty() => Ok(true),
-            Ok(bytes) => Self::load_from_data(disk.clone(), task_id, bytes.to_vec())
-                .await
-                .map(|_| true),
-            Err(crate::heal::DiskError::FileNotFound) => Ok(true),
-            Err(error) => Err(error.into()),
-        }
-    }
-
-    async fn block_invalid_snapshot(disk: &DiskStore, task_id: &str) {
-        // This marker is intentionally version-agnostic: an unsupported reader
-        // must stop selector retries until an operator cleans up the snapshot.
-        let blocked_path = Self::blocked_path(task_id);
-        let Ok(path) = path_to_str(&blocked_path) else {
-            return;
-        };
-        let result = EcstoreDiskAPI::compare_and_update_file(
-            disk.as_ref(),
-            RUSTFS_META_BUCKET,
-            path,
-            None,
-            Some(EcstoreDiskBytes::from_static(b"blocked")),
-        )
-        .await;
-        match result {
-            Ok(EcstoreConditionalFileUpdate::Updated | EcstoreConditionalFileUpdate::Mismatch) => {}
-            Ok(EcstoreConditionalFileUpdate::Missing) => warn!(
-                target: "rustfs::heal::resume",
-                event = EVENT_HEAL_CHECKPOINT_STATE,
-                component = LOG_COMPONENT_HEAL,
-                subsystem = LOG_SUBSYSTEM_RESUME,
-                task_id,
-                state = "blocked_marker_write_failed",
-                error = "marker target disappeared",
-                "Heal checkpoint could not persist its blocked marker"
-            ),
-            Err(error) => warn!(
-                target: "rustfs::heal::resume",
-                event = EVENT_HEAL_CHECKPOINT_STATE,
-                component = LOG_COMPONENT_HEAL,
-                subsystem = LOG_SUBSYSTEM_RESUME,
-                task_id,
-                state = "blocked_marker_write_failed",
-                error = %error,
-                "Heal checkpoint could not persist its blocked marker"
-            ),
-        }
-    }
-
     /// create new checkpoint manager
     pub async fn new(disk: DiskStore, task_id: String) -> Result<Self> {
         validate_resume_task_id(&task_id)?;
-        let checkpoint_volume = format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}");
-        if let Err(error) = EcstoreDiskAPI::make_volume(disk.as_ref(), &checkpoint_volume).await
-            && error != crate::heal::DiskError::VolumeExists
-        {
-            return Err(Error::TaskExecutionFailed {
-                message: format!("Failed to create checkpoint volume: {error}"),
-            });
-        }
         let checkpoint = ResumeCheckpoint::new(task_id);
         let manager = Self {
             disk,
             checkpoint: Arc::new(RwLock::new(checkpoint)),
             throttle: Mutex::new(PersistThrottle::new()),
-            save_lock: AsyncMutex::new(()),
-            last_saved: Mutex::new(None),
         };
 
         // save initial checkpoint
@@ -245,7 +140,6 @@ impl CheckpointManager {
                 error = %e,
                 "Heal checkpoint persistence failed"
             );
-            return Err(e);
         }
         Ok(manager)
     }
@@ -254,22 +148,11 @@ impl CheckpointManager {
     pub async fn load_from_disk(disk: DiskStore, task_id: &str) -> Result<Self> {
         validate_resume_task_id(task_id)?;
         let checkpoint_data = Self::read_checkpoint_file(&disk, task_id).await?;
-        Self::load_from_data(disk, task_id, checkpoint_data).await
-    }
-
-    async fn load_from_data(disk: DiskStore, task_id: &str, checkpoint_data: Vec<u8>) -> Result<Self> {
-        validate_resume_task_id(task_id)?;
-        let mut checkpoint: ResumeCheckpoint = match serde_json::from_slice(&checkpoint_data) {
-            Ok(checkpoint) => checkpoint,
-            Err(error) => {
-                Self::block_invalid_snapshot(&disk, task_id).await;
-                return Err(Error::TaskExecutionFailed {
-                    message: format!("Failed to deserialize checkpoint: {error}"),
-                });
-            }
-        };
+        let mut checkpoint: ResumeCheckpoint =
+            serde_json::from_slice(&checkpoint_data).map_err(|e| Error::TaskExecutionFailed {
+                message: format!("Failed to deserialize checkpoint: {e}"),
+            })?;
         if checkpoint.task_id != task_id {
-            Self::block_invalid_snapshot(&disk, task_id).await;
             return Err(Error::TaskExecutionFailed {
                 message: "Resume checkpoint task id does not match filename".to_string(),
             });
@@ -280,7 +163,6 @@ impl CheckpointManager {
         // identities. Discard the stale sets and position, then stamp the
         // current schema so the scan restarts cleanly.
         if checkpoint.schema_version > CURRENT_CHECKPOINT_SCHEMA {
-            Self::block_invalid_snapshot(&disk, task_id).await;
             return Err(Error::TaskExecutionFailed {
                 message: format!(
                     "Checkpoint schema {} is newer than supported schema {CURRENT_CHECKPOINT_SCHEMA}",
@@ -288,43 +170,7 @@ impl CheckpointManager {
                 ),
             });
         }
-
-        if let Some(expected) = checkpoint.integrity_digest.as_deref() {
-            let actual = Self::checkpoint_digest(&Self::serialize_without_digest(&checkpoint)?);
-            if expected != actual {
-                Self::block_invalid_snapshot(&disk, task_id).await;
-                return Err(Error::InvalidCheckpoint(format!(
-                    "Resume checkpoint digest does not match task {task_id}"
-                )));
-            }
-        } else if checkpoint.schema_version >= CURRENT_CHECKPOINT_SCHEMA {
-            Self::block_invalid_snapshot(&disk, task_id).await;
-            return Err(Error::InvalidCheckpoint(format!(
-                "Resume checkpoint digest is missing for task {task_id}"
-            )));
-        } else {
-            let digest_path = Self::digest_path(task_id);
-            let digest_path = path_to_str(&digest_path)?;
-            match HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, digest_path).await {
-                Ok(expected) => {
-                    let actual = Self::checkpoint_digest(&checkpoint_data);
-                    if expected.as_ref() != actual.as_bytes() {
-                        Self::block_invalid_snapshot(&disk, task_id).await;
-                        return Err(Error::InvalidCheckpoint(format!(
-                            "Resume checkpoint digest does not match task {task_id}"
-                        )));
-                    }
-                }
-                Err(crate::heal::DiskError::FileNotFound) => {}
-                Err(error) => {
-                    return Err(Error::TaskExecutionFailed {
-                        message: format!("Failed to read checkpoint digest: {error}"),
-                    });
-                }
-            }
-        }
-
-        if checkpoint.schema_version < CHECKPOINT_PER_VERSION_SCHEMA {
+        if checkpoint.schema_version < CURRENT_CHECKPOINT_SCHEMA {
             warn!(
                 target: "rustfs::heal::resume",
                 event = EVENT_HEAL_CHECKPOINT_STATE,
@@ -341,15 +187,13 @@ impl CheckpointManager {
             checkpoint.skipped_objects.clear();
             checkpoint.current_bucket_index = 0;
             checkpoint.current_object_index = 0;
+            checkpoint.schema_version = CURRENT_CHECKPOINT_SCHEMA;
         }
-        checkpoint.schema_version = CURRENT_CHECKPOINT_SCHEMA;
 
         Ok(Self {
             disk,
             checkpoint: Arc::new(RwLock::new(checkpoint)),
             throttle: Mutex::new(PersistThrottle::new()),
-            save_lock: AsyncMutex::new(()),
-            last_saved: Mutex::new(Some(EcstoreDiskBytes::from(checkpoint_data))),
         })
     }
 
@@ -360,7 +204,7 @@ impl CheckpointManager {
         }
         let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"));
         match path_to_str(&file_path) {
-            Ok(path_str) => match HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path_str).await {
+            Ok(path_str) => match disk.read_all(RUSTFS_META_BUCKET, path_str).await {
                 Ok(data) => !data.is_empty(),
                 Err(_) => false,
             },
@@ -448,8 +292,6 @@ impl CheckpointManager {
 
         let checkpoint_file = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"));
         delete_resume_file(&self.disk, &checkpoint_file).await?;
-        delete_resume_file(&self.disk, &Self::digest_path(&task_id)).await?;
-        delete_resume_file(&self.disk, &Self::blocked_path(&task_id)).await?;
 
         debug!(
             target: "rustfs::heal::resume",
@@ -465,130 +307,21 @@ impl CheckpointManager {
 
     /// save checkpoint to disk
     async fn save_checkpoint(&self) -> Result<()> {
-        // Serialize saves and take the snapshot only after acquiring the lock:
-        // a slower writer must not publish a snapshot taken before a newer one.
-        let _save_guard = self.save_lock.lock().await;
-        let checkpoint = self.checkpoint.read().await.clone();
+        let checkpoint = self.checkpoint.read().await;
         validate_resume_task_id(&checkpoint.task_id)?;
-        let unsigned_checkpoint_data = Self::serialize_without_digest(&checkpoint)?;
-        let digest = Self::checkpoint_digest(&unsigned_checkpoint_data);
-        let mut persisted_checkpoint = checkpoint.clone();
-        persisted_checkpoint.integrity_digest = Some(digest);
-        let checkpoint_data =
-            EcstoreDiskBytes::from(serde_json::to_vec(&persisted_checkpoint).map_err(|e| Error::TaskExecutionFailed {
-                message: format!("Failed to serialize checkpoint: {e}"),
-            })?);
+        let checkpoint_data = serde_json::to_vec(&*checkpoint).map_err(|e| Error::TaskExecutionFailed {
+            message: format!("Failed to serialize checkpoint: {e}"),
+        })?;
 
         let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{}_{}", checkpoint.task_id, RESUME_CHECKPOINT_FILE));
 
         let path_str = path_to_str(&file_path)?;
-        let last_saved = self
-            .last_saved
-            .lock()
-            .map_err(|_| Error::TaskExecutionFailed {
-                message: "Checkpoint save state lock is poisoned; refusing to save".to_string(),
-            })?
-            .clone();
-        let update = EcstoreDiskAPI::compare_and_update_file(
-            self.disk.as_ref(),
-            RUSTFS_META_BUCKET,
-            path_str,
-            last_saved.clone(),
-            Some(checkpoint_data.clone()),
-        )
-        .await
-        .map_err(|e| Error::TaskExecutionFailed {
-            message: format!("Failed to save checkpoint: {e}"),
-        })?;
-
-        let expected = match update {
-            EcstoreConditionalFileUpdate::Updated => None,
-            EcstoreConditionalFileUpdate::Missing => {
-                return Err(Error::TaskExecutionFailed {
-                    message: "Checkpoint was removed after this manager saved it; refusing to recreate it".to_string(),
-                });
-            }
-            EcstoreConditionalFileUpdate::Mismatch => {
-                // A healthy manager normally completes the CAS above without
-                // another read or JSON parse. Inspect only after a mismatch so
-                // corruption and future schemas cannot be overwritten blindly.
-                let existing = match HealDiskExt::read_all(self.disk.as_ref(), RUSTFS_META_BUCKET, path_str).await {
-                    Ok(existing) => existing,
-                    Err(crate::heal::DiskError::FileNotFound) => {
-                        return Err(Error::TaskExecutionFailed {
-                            message: "Checkpoint was removed after this manager saved it; refusing to recreate it".to_string(),
-                        });
-                    }
-                    Err(error) => {
-                        return Err(Error::TaskExecutionFailed {
-                            message: format!("Failed to inspect checkpoint after CAS mismatch: {error}"),
-                        });
-                    }
-                };
-
-                if existing.is_empty() && last_saved.is_none() {
-                    Some(existing)
-                } else {
-                    let current: ResumeCheckpoint = match serde_json::from_slice(&existing) {
-                        Ok(current) => current,
-                        Err(error) => {
-                            Self::block_invalid_snapshot(&self.disk, &checkpoint.task_id).await;
-                            return Err(Error::TaskExecutionFailed {
-                                message: format!("Existing checkpoint is corrupt: {error}"),
-                            });
-                        }
-                    };
-                    if current.task_id != checkpoint.task_id {
-                        Self::block_invalid_snapshot(&self.disk, &checkpoint.task_id).await;
-                        return Err(Error::TaskExecutionFailed {
-                            message: "Existing checkpoint task id does not match filename".to_string(),
-                        });
-                    }
-                    if current.schema_version > CURRENT_CHECKPOINT_SCHEMA {
-                        Self::block_invalid_snapshot(&self.disk, &checkpoint.task_id).await;
-                        return Err(Error::TaskExecutionFailed {
-                            message: format!(
-                                "Existing checkpoint schema {} is newer than supported schema {CURRENT_CHECKPOINT_SCHEMA}",
-                                current.schema_version
-                            ),
-                        });
-                    }
-                    if last_saved.as_ref().is_none_or(|saved| saved.as_ref() != existing.as_ref()) {
-                        return Err(Error::TaskExecutionFailed {
-                            message: "Checkpoint changed since this manager loaded it; refusing to overwrite newer progress"
-                                .to_string(),
-                        });
-                    }
-                    Some(existing)
-                }
-            }
-        };
-
-        if let Some(expected) = expected {
-            match EcstoreDiskAPI::compare_and_update_file(
-                self.disk.as_ref(),
-                RUSTFS_META_BUCKET,
-                path_str,
-                Some(expected),
-                Some(checkpoint_data.clone()),
-            )
+        self.disk
+            .write_all(RUSTFS_META_BUCKET, path_str, checkpoint_data.into())
             .await
             .map_err(|e| Error::TaskExecutionFailed {
-                message: format!("Failed to save checkpoint after CAS mismatch: {e}"),
-            })? {
-                EcstoreConditionalFileUpdate::Updated => {}
-                EcstoreConditionalFileUpdate::Missing | EcstoreConditionalFileUpdate::Mismatch => {
-                    return Err(Error::TaskExecutionFailed {
-                        message: "Checkpoint changed while saving; refusing to overwrite newer progress".to_string(),
-                    });
-                }
-            }
-        }
-
-        let mut last_saved = self.last_saved.lock().map_err(|_| Error::TaskExecutionFailed {
-            message: "Checkpoint save state lock is poisoned after save".to_string(),
-        })?;
-        *last_saved = Some(checkpoint_data);
+                message: format!("Failed to save checkpoint: {e}"),
+            })?;
 
         debug!(
             target: "rustfs::heal::resume",
@@ -608,38 +341,11 @@ impl CheckpointManager {
         let file_path = Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"));
 
         let path_str = path_to_str(&file_path)?;
-        HealDiskExt::read_all(disk.as_ref(), RUSTFS_META_BUCKET, path_str)
+        disk.read_all(RUSTFS_META_BUCKET, path_str)
             .await
             .map(|bytes| bytes.to_vec())
             .map_err(|e| Error::TaskExecutionFailed {
                 message: format!("Failed to read checkpoint file: {e}"),
             })
     }
-
-    fn serialize_without_digest(checkpoint: &ResumeCheckpoint) -> Result<Vec<u8>> {
-        let mut unsigned = checkpoint.clone();
-        unsigned.integrity_digest = None;
-        let mut value = serde_json::to_value(&unsigned).map_err(|e| Error::TaskExecutionFailed {
-            message: format!("Failed to serialize checkpoint: {e}"),
-        })?;
-        for field in ["processed_objects", "failed_objects", "skipped_objects"] {
-            let Some(values) = value.get_mut(field).and_then(serde_json::Value::as_array_mut) else {
-                return Err(Error::TaskExecutionFailed {
-                    message: format!("Failed to canonicalize checkpoint field: {field}"),
-                });
-            };
-            values.sort_by(|left, right| left.as_str().cmp(&right.as_str()));
-        }
-        serde_json::to_vec(&value).map_err(|e| Error::TaskExecutionFailed {
-            message: format!("Failed to serialize checkpoint: {e}"),
-        })
-    }
-
-    fn checkpoint_digest(checkpoint_data: &[u8]) -> String {
-        base64::engine::general_purpose::STANDARD.encode(Sha256::digest(checkpoint_data))
-    }
-
-    fn digest_path(task_id: &str) -> std::path::PathBuf {
-        Path::new(BUCKET_META_PREFIX).join(format!("{task_id}_{RESUME_CHECKPOINT_DIGEST_FILE}"))
-    }
 }

crates/heal/src/heal/resume/tests.rs

@@ -1600,29 +1600,6 @@ async fn test_checkpoint_schema_v4_discarded_on_load() {
     temp_dir.close().expect("remove schema test directory");
 }
 
-#[tokio::test]
-async fn unsigned_previous_checkpoint_schema_preserves_progress() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let mut legacy = ResumeCheckpoint::new(task_id.clone());
-    legacy.schema_version = CURRENT_CHECKPOINT_SCHEMA - 1;
-    legacy.update_position(2, 500);
-    legacy.add_processed_object("object".to_string());
-    legacy.integrity_digest = None;
-    let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
-    disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, serde_json::to_vec(&legacy).unwrap().into())
-        .await
-        .expect("write previous-schema checkpoint");
-
-    let manager = CheckpointManager::load_from_disk(disk, &task_id).await.unwrap();
-    let checkpoint = manager.get_checkpoint().await;
-    assert_eq!(checkpoint.schema_version, CURRENT_CHECKPOINT_SCHEMA);
-    assert_eq!(checkpoint.current_bucket_index, 2);
-    assert_eq!(checkpoint.current_object_index, 500);
-    assert!(checkpoint.processed_objects.contains("object"));
-    temp_dir.close().unwrap();
-}
-
 #[tokio::test]
 async fn current_normal_resume_schema_preserves_progress() {
     let (temp_dir, disk) = schema_test_disk().await;
@@ -1698,369 +1675,6 @@ async fn future_resume_and_checkpoint_schemas_are_rejected() {
     temp_dir.close().expect("remove schema test directory");
 }
 
-#[tokio::test]
-async fn checkpoint_save_does_not_replace_a_non_empty_truncated_snapshot() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
-        .await
-        .expect("create checkpoint manager");
-    let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
-    let truncated = b"{\"schema_version\":5,\"task_id\":";
-    disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, truncated.as_slice().into())
-        .await
-        .expect("write truncated checkpoint fixture");
-
-    let error = manager
-        .update_position(2, 7)
-        .await
-        .expect_err("a truncated checkpoint must fail closed during save");
-    assert!(error.to_string().contains("Existing checkpoint is corrupt"));
-    assert_eq!(
-        disk.read_all(RUSTFS_META_BUCKET, &checkpoint_path)
-            .await
-            .expect("read truncated checkpoint fixture"),
-        truncated.as_slice()
-    );
-    assert!(CheckpointManager::is_blocked(&disk, &task_id).await);
-    temp_dir.close().expect("remove checkpoint save test directory");
-}
-
-#[tokio::test]
-async fn checkpoint_save_does_not_replace_a_future_schema_snapshot() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
-        .await
-        .expect("create checkpoint manager");
-    let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
-    let mut future = ResumeCheckpoint::new(task_id.clone());
-    future.schema_version = CURRENT_CHECKPOINT_SCHEMA + 1;
-    let future_bytes = serde_json::to_vec(&future).expect("serialize future checkpoint fixture");
-    disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, future_bytes.clone().into())
-        .await
-        .expect("write future checkpoint fixture");
-
-    let error = manager
-        .update_position(2, 7)
-        .await
-        .expect_err("a future schema must fail closed during save");
-    assert!(error.to_string().contains("Existing checkpoint schema"));
-    assert_eq!(
-        disk.read_all(RUSTFS_META_BUCKET, &checkpoint_path)
-            .await
-            .expect("read future checkpoint fixture"),
-        future_bytes
-    );
-    assert!(CheckpointManager::is_blocked(&disk, &task_id).await);
-    temp_dir.close().expect("remove future schema test directory");
-}
-
-#[tokio::test]
-async fn checkpoint_digest_rejects_same_length_progress_tampering() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
-        .await
-        .expect("create checkpoint manager");
-    manager
-        .add_processed_object("victim-a".to_string())
-        .await
-        .expect("persist checkpoint progress");
-    manager.update_position(1, 1).await.expect("flush checkpoint progress");
-    let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
-    let original = disk
-        .read_all(RUSTFS_META_BUCKET, &checkpoint_path)
-        .await
-        .expect("read checkpoint fixture");
-    let tampered = original
-        .windows(b"victim-a".len())
-        .position(|window| window == b"victim-a")
-        .map(|index| {
-            let mut bytes = original.to_vec();
-            bytes[index..index + b"victim-a".len()].copy_from_slice(b"victim-b");
-            bytes
-        })
-        .expect("checkpoint should contain the processed object");
-    disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, tampered.into())
-        .await
-        .expect("write tampered checkpoint fixture");
-
-    assert!(CheckpointManager::load_from_disk(disk.clone(), &task_id).await.is_err());
-    assert!(CheckpointManager::is_blocked(&disk, &task_id).await);
-    temp_dir.close().expect("remove digest test directory");
-}
-
-#[tokio::test]
-async fn checkpoint_integrity_survives_missing_legacy_sidecar() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let manager = CheckpointManager::new(disk.clone(), task_id.clone()).await.unwrap();
-    manager.update_position(2, 9).await.unwrap();
-
-    let digest_path = format!("{BUCKET_META_PREFIX}/{task_id}_ahm_checkpoint.sha256");
-    delete_resume_file(&disk, Path::new(&digest_path)).await.unwrap();
-
-    let restored = CheckpointManager::load_from_disk(disk, &task_id).await.unwrap();
-    let checkpoint = restored.get_checkpoint().await;
-    assert_eq!(checkpoint.current_bucket_index, 2);
-    assert_eq!(checkpoint.current_object_index, 9);
-    assert!(checkpoint.integrity_digest.is_some());
-    temp_dir.close().unwrap();
-}
-
-#[tokio::test]
-async fn checkpoint_integrity_survives_multi_object_reload() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let manager = CheckpointManager::new(disk.clone(), task_id.clone()).await.unwrap();
-    for index in 0..32 {
-        manager.add_processed_object(format!("processed-{index}")).await.unwrap();
-        manager.add_failed_object(format!("failed-{index}")).await.unwrap();
-        manager.add_skipped_object(format!("skipped-{index}")).await.unwrap();
-    }
-    manager.update_position(2, 9).await.unwrap();
-
-    CheckpointManager::load_from_disk(disk, &task_id)
-        .await
-        .expect("a healthy multi-object checkpoint must survive reload");
-    temp_dir.close().unwrap();
-}
-
-#[tokio::test]
-async fn checkpoint_integrity_rejects_a_removed_embedded_digest() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let manager = CheckpointManager::new(disk.clone(), task_id.clone()).await.unwrap();
-    manager.update_position(2, 9).await.unwrap();
-
-    let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
-    let bytes = disk
-        .read_all(RUSTFS_META_BUCKET, &checkpoint_path)
-        .await
-        .expect("read checkpoint fixture");
-    let mut value: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
-    value["current_object_index"] = serde_json::json!(10);
-    value.as_object_mut().unwrap().remove("integrity_digest");
-    disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, serde_json::to_vec(&value).unwrap().into())
-        .await
-        .expect("write tampered checkpoint fixture");
-
-    assert!(
-        CheckpointManager::load_from_disk(disk.clone(), &task_id).await.is_err(),
-        "a current checkpoint without its embedded digest must fail closed"
-    );
-    assert!(CheckpointManager::is_blocked(&disk, &task_id).await);
-    temp_dir.close().unwrap();
-}
-
-#[tokio::test]
-async fn new_checkpoint_manager_rebuilds_an_empty_snapshot() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
-    disk.write_all(RUSTFS_META_BUCKET, &checkpoint_path, EcstoreDiskBytes::new())
-        .await
-        .expect("write empty checkpoint fixture");
-
-    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
-        .await
-        .expect("a new manager must rebuild an empty checkpoint");
-    manager
-        .update_position(3, 11)
-        .await
-        .expect("rebuilt checkpoint must remain writable");
-    assert!(CheckpointManager::has_checkpoint(&disk, &task_id).await);
-    temp_dir.close().expect("remove empty checkpoint test directory");
-}
-
-#[tokio::test]
-async fn deleted_checkpoint_is_not_recreated_by_an_old_manager() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
-        .await
-        .expect("create checkpoint manager");
-    manager.cleanup().await.expect("delete checkpoint fixture");
-
-    let error = manager
-        .update_position(1, 2)
-        .await
-        .expect_err("an old manager must not resurrect a deleted checkpoint");
-    assert!(error.to_string().contains("removed after this manager saved it"));
-    assert!(!CheckpointManager::has_checkpoint(&disk, &task_id).await);
-    temp_dir.close().expect("remove deleted checkpoint test directory");
-}
-
-#[cfg(unix)]
-#[tokio::test]
-async fn checkpoint_cleanup_leaves_no_task_specific_lock_artifact() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
-        .await
-        .expect("create checkpoint manager");
-    let lock_path = Path::new(BUCKET_META_PREFIX)
-        .join(format!("{task_id}_{RESUME_CHECKPOINT_FILE}"))
-        .with_extension("rustfs-cas.lock");
-    let lock_path = temp_dir.path().join(RUSTFS_META_BUCKET).join(lock_path);
-
-    manager.cleanup().await.expect("delete checkpoint fixture");
-
-    assert!(
-        !lock_path.exists(),
-        "successful checkpoint cleanup must not leave a task-specific lock artifact"
-    );
-}
-
-#[tokio::test]
-async fn an_empty_blocked_marker_still_blocks_resume_selection() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let manager = CheckpointManager::new(disk.clone(), task_id.clone())
-        .await
-        .expect("create checkpoint manager");
-    let blocked_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}");
-    disk.write_all(RUSTFS_META_BUCKET, &blocked_path, EcstoreDiskBytes::new())
-        .await
-        .expect("write empty blocked marker fixture");
-
-    assert!(CheckpointManager::is_blocked(&disk, &task_id).await);
-    assert!(CheckpointManager::is_resumable(&disk, &task_id).await.is_err());
-    // Recovery requires replacing/cleaning the snapshot, then removing the
-    // marker; ordinary selector retries are intentionally not an unlock path.
-    manager.cleanup().await.expect("clean blocked checkpoint");
-    assert!(!CheckpointManager::is_blocked(&disk, &task_id).await);
-    temp_dir.close().expect("remove empty blocked marker test directory");
-}
-
-#[tokio::test]
-async fn resumable_selector_skips_healthy_tasks_with_blocked_markers() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let tasks = [
-        (ResumeUtils::generate_task_id(), EcstoreDiskBytes::new()),
-        (ResumeUtils::generate_task_id(), EcstoreDiskBytes::from_static(b"blocked")),
-    ];
-    for (task_id, marker) in &tasks {
-        ResumeManager::new(
-            disk.clone(),
-            task_id.clone(),
-            "erasure_set".to_string(),
-            "pool_0_set_0".to_string(),
-            vec!["bucket".to_string()],
-        )
-        .await
-        .expect("create healthy resume state");
-        CheckpointManager::new(disk.clone(), task_id.clone())
-            .await
-            .expect("create healthy checkpoint");
-        let checkpoint_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
-        let checkpoint_bytes = disk
-            .read_all(RUSTFS_META_BUCKET, &checkpoint_path)
-            .await
-            .expect("read healthy checkpoint before blocking");
-        let marker_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}");
-        disk.write_all(RUSTFS_META_BUCKET, &marker_path, marker.clone())
-            .await
-            .expect("write blocked marker");
-
-        assert!(ResumeUtils::get_resumable_tasks(&disk).await.is_err());
-        assert_eq!(
-            disk.read_all(RUSTFS_META_BUCKET, &checkpoint_path)
-                .await
-                .expect("read healthy checkpoint after blocking"),
-            checkpoint_bytes
-        );
-    }
-    temp_dir.close().expect("remove blocked selector test directory");
-}
-
-#[tokio::test]
-async fn stale_checkpoint_manager_cannot_overwrite_newer_progress() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let task_id = ResumeUtils::generate_task_id();
-    let first = CheckpointManager::new(disk.clone(), task_id.clone())
-        .await
-        .expect("create first checkpoint manager");
-    let second = CheckpointManager::load_from_disk(disk.clone(), &task_id)
-        .await
-        .expect("load second checkpoint manager");
-
-    second
-        .update_position(4, 20)
-        .await
-        .expect("persist newer checkpoint progress");
-    let error = first
-        .update_position(1, 3)
-        .await
-        .expect_err("stale checkpoint manager must not overwrite newer progress");
-    assert!(error.to_string().contains("newer progress"));
-
-    let persisted = CheckpointManager::load_from_disk(disk.clone(), &task_id)
-        .await
-        .expect("load newer checkpoint progress")
-        .get_checkpoint()
-        .await;
-    assert_eq!(persisted.current_bucket_index, 4);
-    assert_eq!(persisted.current_object_index, 20);
-    temp_dir.close().expect("remove stale manager test directory");
-}
-
-#[tokio::test]
-async fn resumable_selector_isolates_future_and_corrupt_checkpoints() {
-    let (temp_dir, disk) = schema_test_disk().await;
-    let future_task = ResumeUtils::generate_task_id();
-    let corrupt_task = ResumeUtils::generate_task_id();
-    for task_id in [&future_task, &corrupt_task] {
-        ResumeManager::new(
-            disk.clone(),
-            task_id.to_string(),
-            "erasure_set".to_string(),
-            "pool_0_set_0".to_string(),
-            vec!["bucket".to_string()],
-        )
-        .await
-        .expect("create resumable state fixture");
-    }
-
-    let future_path = format!("{BUCKET_META_PREFIX}/{future_task}_{RESUME_CHECKPOINT_FILE}");
-    let mut future = ResumeCheckpoint::new(future_task.clone());
-    future.schema_version = CURRENT_CHECKPOINT_SCHEMA + 1;
-    let future_bytes = serde_json::to_vec(&future).expect("serialize future checkpoint fixture");
-    disk.write_all(RUSTFS_META_BUCKET, &future_path, future_bytes.clone().into())
-        .await
-        .expect("write future checkpoint fixture");
-    let corrupt_path = format!("{BUCKET_META_PREFIX}/{corrupt_task}_{RESUME_CHECKPOINT_FILE}");
-    let corrupt_bytes = b"{truncated";
-    disk.write_all(RUSTFS_META_BUCKET, &corrupt_path, corrupt_bytes.as_slice().into())
-        .await
-        .expect("write corrupt checkpoint fixture");
-
-    assert!(CheckpointManager::is_resumable(&disk, &future_task).await.is_err());
-    assert!(CheckpointManager::is_resumable(&disk, &corrupt_task).await.is_err());
-    assert!(ResumeUtils::get_resumable_tasks(&disk).await.is_err());
-    for (task_id, path, bytes) in [
-        (&future_task, future_path, future_bytes),
-        (&corrupt_task, corrupt_path, corrupt_bytes.to_vec()),
-    ] {
-        assert_eq!(
-            disk.read_all(RUSTFS_META_BUCKET, &path)
-                .await
-                .expect("read isolated checkpoint bytes"),
-            bytes
-        );
-        let blocked_path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_BLOCKED_FILE}");
-        assert!(
-            !disk
-                .read_all(RUSTFS_META_BUCKET, &blocked_path)
-                .await
-                .expect("read checkpoint blocked marker")
-                .is_empty()
-        );
-    }
-    temp_dir.close().expect("remove selector isolation test directory");
-}
-
 #[test]
 fn test_persist_throttle_batches_until_threshold() {
     let mut throttle = PersistThrottle::new();

crates/heal/src/heal/resume/utils.rs

@@ -21,7 +21,7 @@ use uuid::Uuid;
 use super::super::{BUCKET_META_PREFIX, DiskError, DiskStore, HealDiskExt as _, RUSTFS_META_BUCKET};
 use super::replacement::{ReplacementPhase, ReplacementRecoveryRecord};
 use super::{
-    CheckpointManager, EVENT_HEAL_RESUME_STATE, LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, REPLACEMENT_COMPLETION_PROOF_FILE,
+    EVENT_HEAL_RESUME_STATE, LOG_COMPONENT_HEAL, LOG_SUBSYSTEM_RESUME, REPLACEMENT_COMPLETION_PROOF_FILE,
     REPLACEMENT_INTENT_FILE, RESUME_STATE_FILE, ResumeManager, ResumeStateFile, is_replacement_intent, path_to_str,
     replacement_recovery_corruption_for_state_load, replacement_recovery_dir, validate_resume_task_id,
 };
@@ -67,7 +67,6 @@ impl ResumeUtils {
                 // Extract task ID from filename: {task_id}_ahm_resume_state.json
                 if let Some(task_id) = entry.strip_suffix(&format!("_{RESUME_STATE_FILE}"))
                     && validate_resume_task_id(task_id).is_ok()
-                    && CheckpointManager::is_resumable(disk, task_id).await?
                 {
                     task_ids.push(task_id.to_string());
                 }

crates/madmin/src/metrics.rs

@@ -689,6 +689,14 @@ pub struct ScannerMetrics {
     pub cycle_max_objects: u64,
     #[serde(rename = "cycle_max_directories", default)]
     pub cycle_max_directories: u64,
+    #[serde(rename = "cycle_timeout_total", default)]
+    pub cycle_timeout_total: u64,
+    #[serde(rename = "cycle_recovery_required_total", default)]
+    pub cycle_recovery_required_total: u64,
+    #[serde(rename = "cycle_last_progress_age", default)]
+    pub cycle_last_progress_age: u64,
+    #[serde(rename = "leader_lease_without_progress", default)]
+    pub leader_lease_without_progress: bool,
     #[serde(rename = "bitrot_cycle_enabled", default)]
     pub bitrot_cycle_enabled: bool,
     #[serde(rename = "bitrot_cycle_seconds", default)]
@@ -764,6 +772,8 @@ impl ScannerMetrics {
             self.cycle_max_duration_seconds = other.cycle_max_duration_seconds;
             self.cycle_max_objects = other.cycle_max_objects;
             self.cycle_max_directories = other.cycle_max_directories;
+            self.cycle_last_progress_age = other.cycle_last_progress_age;
+            self.leader_lease_without_progress = other.leader_lease_without_progress;
             self.bitrot_cycle_enabled = other.bitrot_cycle_enabled;
             self.bitrot_cycle_seconds = other.bitrot_cycle_seconds;
         }
@@ -857,6 +867,12 @@ impl ScannerMetrics {
             .saturating_add(other.last_cycle_replication_checks);
         self.last_cycle_usage_saves = self.last_cycle_usage_saves.saturating_add(other.last_cycle_usage_saves);
         self.failed_cycles = self.failed_cycles.saturating_add(other.failed_cycles);
+        self.cycle_timeout_total = self.cycle_timeout_total.saturating_add(other.cycle_timeout_total);
+        self.cycle_recovery_required_total = self
+            .cycle_recovery_required_total
+            .saturating_add(other.cycle_recovery_required_total);
+        self.cycle_last_progress_age = self.cycle_last_progress_age.max(other.cycle_last_progress_age);
+        self.leader_lease_without_progress |= other.leader_lease_without_progress;
         self.superseded_cycles = self.superseded_cycles.saturating_add(other.superseded_cycles);
         self.partial_cycles_unknown = self.partial_cycles_unknown.saturating_add(other.partial_cycles_unknown);
         self.partial_cycles_runtime = self.partial_cycles_runtime.saturating_add(other.partial_cycles_runtime);

crates/scanner/src/data_usage_define.rs

@@ -29,7 +29,8 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
 pub use rustfs_data_usage::{
     AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
     DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
-    PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path, prefix_usage_in_cache,
+    PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, UNKNOWN_TIER, UnknownTierStats,
+    hash_path, prefix_usage_in_cache,
 };
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
 use tokio::time::{Duration, Instant, sleep, timeout};
@@ -205,7 +206,8 @@ impl ScannerSizeSummaryExt for SizeSummary {
             self.versions = self.versions.saturating_add(1);
         }
 
-        let size = usize::try_from(size.max(0)).unwrap_or(usize::MAX);
+        let logical_size = size.max(0);
+        let size = usize::try_from(logical_size).unwrap_or(usize::MAX);
         self.total_size = self.total_size.saturating_add(size);
 
         if oi.transitioned_object.free_version {
@@ -217,12 +219,34 @@ impl ScannerSizeSummaryExt for SizeSummary {
             tier = oi.transitioned_object.tier.clone();
         }
 
-        if let Some(tier_stats) = self.tier_stats.get_mut(&tier) {
-            *tier_stats = tier_stats.add(&TierStats {
-                total_size: u64::try_from(oi.size).unwrap_or(0),
-                num_versions: 1,
-                num_objects: u64::from(oi.is_latest),
-            });
+        let builtin_tier = tier == storageclass::STANDARD || tier == storageclass::RRS;
+        let tier_registry_is_empty =
+            self.tier_stats.is_empty() || (self.tier_stats.len() == 1 && self.tier_stats.contains_key(UNKNOWN_TIER));
+        let known_tier = tier != UNKNOWN_TIER && (builtin_tier || self.tier_stats.contains_key(&tier));
+
+        // With no configured tier, retain the historical empty-map shape for
+        // ordinary STANDARD/RRS objects. A non-built-in key is still an
+        // observable unknown and must create only the fixed bucket.
+        if tier_registry_is_empty && known_tier {
+            return;
+        }
+
+        let accounting_key = if known_tier { tier.clone() } else { UNKNOWN_TIER.to_string() };
+        let tier_stats = self.tier_stats.entry(accounting_key).or_default();
+        let physical_size = u64::try_from(oi.size.max(0)).unwrap_or(0);
+        *tier_stats = tier_stats.add(&TierStats {
+            total_size: physical_size,
+            num_versions: 1,
+            num_objects: u64::from(oi.is_latest),
+        });
+        if !known_tier {
+            self.unknown_tier_stats.record_dimensions(
+                &tier,
+                u64::try_from(logical_size).unwrap_or(u64::MAX),
+                physical_size,
+                1,
+                u64::from(oi.is_latest),
+            );
         }
     }
 }
@@ -344,6 +368,10 @@ pub struct DataUsageCacheInfo {
     pub scan_plan_digest: Option<DataUsageScanPlanDigest>,
     #[serde(default)]
     pub cache_key_format: u16,
+    /// Registry generation used for the completed/partial scan. This is
+    /// process-local audit data; older cache writers omit it.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub tier_registry_generation: Option<u64>,
 }
 
 impl Serialize for DataUsageCacheInfo {
@@ -353,7 +381,7 @@ impl Serialize for DataUsageCacheInfo {
     {
         // Keep this metadata map-encoded so older readers can ignore fields
         // appended by newer scanner versions during rolling upgrades.
-        let mut state = serializer.serialize_map(Some(16))?;
+        let mut state = serializer.serialize_map(Some(17))?;
         state.serialize_entry("name", &self.name)?;
         state.serialize_entry("next_cycle", &self.next_cycle)?;
         state.serialize_entry("leader_epoch", &self.leader_epoch)?;
@@ -370,6 +398,7 @@ impl Serialize for DataUsageCacheInfo {
         state.serialize_entry("snapshot_complete", &self.snapshot_complete)?;
         state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?;
         state.serialize_entry("cache_key_format", &self.cache_key_format)?;
+        state.serialize_entry("tier_registry_generation", &self.tier_registry_generation)?;
         state.end()
     }
 }
@@ -390,6 +419,17 @@ pub(crate) enum DataUsageCachePrepareOutcome {
 }
 
 impl DataUsageCache {
+    /// Reconcile tier keys loaded from an older cache against the registry
+    /// frozen for this scan. New metadata is already routed through
+    /// `UNKNOWN_TIER`; this pass handles retired keys that predate that rule.
+    pub(crate) fn fold_retired_tiers(&mut self, tier_names: &[String]) {
+        for entry in self.cache.values_mut() {
+            if let Some(tiers) = entry.all_tier_stats.as_mut() {
+                tiers.fold_unknown_tiers(tier_names.iter().map(String::as_str));
+            }
+        }
+    }
+
     /// Prefix-level usage query over this (writer-side) cache; see
     /// [`prefix_usage_in_cache`] for the semantics
     /// (rustfs/backlog#1872).
@@ -875,6 +915,7 @@ impl DataUsageCache {
             delete_markers_total_count: flat.delete_markers as u64,
             objects_total_size: flat.size as u64,
             tier_stats: flat.all_tier_stats.filter(|tiers| !tiers.is_empty()),
+            unknown_tier_stats: flat.unknown_tier_stats.filter(|stats| !stats.is_empty()),
             buckets_count: u64::try_from(buckets.len()).unwrap_or(u64::MAX),
             buckets_usage,
             ..Default::default()

crates/scanner/src/data_usage_define/tests.rs

@@ -573,7 +573,6 @@ fn size_summary_add_saturates_all_usage_counters() {
             failed_count: usize::MAX,
         },
     );
-
     let mut increment = SizeSummary {
         total_size: 1,
         versions: 1,
@@ -588,6 +587,24 @@ fn size_summary_add_saturates_all_usage_counters() {
         failed_count: 1,
         ..Default::default()
     };
+    summary.tier_stats.insert(
+        UNKNOWN_TIER.to_string(),
+        TierStats {
+            total_size: u64::MAX,
+            num_versions: u64::MAX,
+            num_objects: u64::MAX,
+        },
+    );
+    increment.tier_stats.insert(
+        UNKNOWN_TIER.to_string(),
+        TierStats {
+            total_size: 1,
+            num_versions: 1,
+            num_objects: 1,
+        },
+    );
+    increment.unknown_tier_stats.unknown_bytes = 1;
+    increment.unknown_tier_stats.unknown_physical_bytes = 1;
     increment.repl_target_stats.insert(
         target.clone(),
         ReplTargetSizeSummary {
@@ -624,6 +641,8 @@ fn size_summary_add_saturates_all_usage_counters() {
     assert_eq!(target_summary.failed_size, i64::MAX);
     assert_eq!(target_summary.pending_count, usize::MAX);
     assert_eq!(target_summary.failed_count, usize::MAX);
+    assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, u64::MAX);
+    assert_eq!(summary.unknown_tier_stats.unknown_bytes, 1);
 }
 
 #[test]
@@ -673,6 +692,162 @@ fn size_summary_actions_accounting_accumulates_tier_stats() {
     );
 }
 
+#[test]
+fn unknown_tier_is_bounded_and_accounted() {
+    let mut summary = SizeSummary::new();
+    summary.tier_stats.insert("WARM".to_string(), TierStats::default());
+    let object = ObjectInfo {
+        storage_class: Some("retired-tier".to_string()),
+        size: 11,
+        is_latest: true,
+        ..Default::default()
+    };
+
+    summary.actions_accounting(&object, 11, 11);
+
+    assert_eq!(summary.tier_stats.len(), 2);
+    assert_eq!(summary.tier_stats.get(UNKNOWN_TIER).map(|stats| stats.total_size), Some(11));
+    assert_eq!(summary.unknown_tier_stats.unknown_bytes, 11);
+    assert_eq!(summary.unknown_tier_stats.unknown_physical_bytes, 11);
+    assert_eq!(summary.unknown_tier_stats.unknown_objects, 1);
+    assert!(
+        summary
+            .unknown_tier_stats
+            .diagnostics
+            .iter()
+            .all(|entry| !entry.contains("retired"))
+    );
+}
+
+#[test]
+fn unknown_tier_is_accounted_when_no_remote_tier_is_configured() {
+    let mut summary = SizeSummary::new();
+    let object = ObjectInfo {
+        storage_class: Some("retired-tier".to_string()),
+        size: 3,
+        is_latest: true,
+        ..Default::default()
+    };
+
+    summary.actions_accounting(&object, 9, 9);
+
+    assert_eq!(summary.tier_stats.len(), 1);
+    assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, 3);
+    assert_eq!(summary.unknown_tier_stats.unknown_bytes, 9);
+    assert_eq!(summary.unknown_tier_stats.unknown_physical_bytes, 3);
+
+    let standard = ObjectInfo {
+        storage_class: Some(storageclass::STANDARD.to_string()),
+        size: 4,
+        is_latest: true,
+        ..Default::default()
+    };
+    summary.actions_accounting(&standard, 4, 4);
+    assert_eq!(summary.tier_stats.len(), 1, "built-ins preserve the no-tier map shape");
+}
+
+#[test]
+fn million_unique_tier_keys_do_not_grow_stats_map() {
+    let mut summary = SizeSummary::new();
+    summary.tier_stats.insert("WARM".to_string(), TierStats::default());
+    for index in 0..1_000_000_u64 {
+        let object = ObjectInfo {
+            storage_class: Some(format!("untrusted-tier-{index}")),
+            size: 1,
+            ..Default::default()
+        };
+        summary.actions_accounting(&object, 1, 1);
+    }
+
+    assert_eq!(summary.tier_stats.len(), 2);
+    assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, 1_000_000);
+    assert_eq!(summary.unknown_tier_stats.unknown_bytes, 1_000_000);
+}
+
+#[test]
+fn unknown_tier_never_triggers_transition() {
+    let mut summary = SizeSummary::new();
+    summary.tier_stats.insert("WARM".to_string(), TierStats::default());
+    let mut object = ObjectInfo {
+        storage_class: Some("removed-tier".to_string()),
+        size: 7,
+        ..Default::default()
+    };
+    object.transitioned_object.status = TRANSITION_COMPLETE.to_string();
+    object.transitioned_object.tier = "removed-tier".to_string();
+
+    summary.actions_accounting(&object, 7, 7);
+
+    assert_eq!(summary.tier_stats.get("removed-tier"), None);
+    assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, 7);
+}
+
+#[test]
+fn removed_tier_survives_restart_as_unknown() {
+    let mut summary = SizeSummary::new();
+    summary.tier_stats.insert("COLD".to_string(), TierStats::default());
+    let object = ObjectInfo {
+        storage_class: Some("retired-tier".to_string()),
+        size: 5,
+        ..Default::default()
+    };
+    summary.actions_accounting(&object, 5, 5);
+    let mut entry = DataUsageEntry::default();
+    entry.add_tier_sizes(&summary.tier_stats);
+    entry.add_unknown_tier_stats(&summary.unknown_tier_stats);
+    let encoded = rmp_serde::to_vec(&entry).expect("entry should encode");
+    let restored: DataUsageEntry = rmp_serde::from_slice(&encoded).expect("entry should decode");
+
+    assert_eq!(restored.all_tier_stats.expect("tier stats persisted").tiers[UNKNOWN_TIER].total_size, 5);
+    assert_eq!(restored.unknown_tier_stats.expect("unknown stats persisted").unknown_bytes, 5);
+}
+
+#[test]
+fn tier_registry_refresh_does_not_mix_cycle_generations() {
+    let first = crate::TierRegistrySnapshot {
+        generation: 1,
+        names: Arc::from(["WARM".to_string()]),
+        refresh_failed: false,
+    };
+    let second = crate::TierRegistrySnapshot {
+        generation: 2,
+        names: Arc::from(["COLD".to_string()]),
+        refresh_failed: false,
+    };
+    assert_ne!(first.generation, second.generation);
+    assert_eq!(first.names.as_ref(), ["WARM".to_string()]);
+    assert_eq!(second.names.as_ref(), ["COLD".to_string()]);
+    assert!(first.refreshed(Err(())).refresh_failed);
+    assert!(!second.refreshed(Ok(Arc::from(["HOT".to_string()]))).refresh_failed);
+    assert_eq!(first.refreshed(Err(())).generation, first.generation);
+    assert_eq!(first.refreshed(Err(())).names, first.names);
+}
+
+#[test]
+fn unknown_tier_counter_uses_checked_arithmetic() {
+    let max = TierStats {
+        total_size: u64::MAX,
+        num_versions: u64::MAX,
+        num_objects: u64::MAX,
+    };
+    assert!(max.checked_add(&TierStats::default()).is_some());
+    assert!(
+        max.checked_add(&TierStats {
+            total_size: 1,
+            ..Default::default()
+        })
+        .is_none()
+    );
+
+    let mut unknown = UnknownTierStats {
+        unknown_bytes: u64::MAX,
+        ..Default::default()
+    };
+    unknown.record("overflow", 1, 1, 1);
+    assert_eq!(unknown.unknown_bytes, u64::MAX);
+    assert_eq!(unknown.unknown_objects, 1);
+}
+
 #[test]
 fn test_data_usage_entry_merge_sums_failed_objects() {
     let mut left = DataUsageEntry {
@@ -926,7 +1101,7 @@ const USAGE_CACHE_WIRE_FIXTURE: &[u8] = &[
     0xdc, 0x00, 0x20, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03,
     0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0xb0, 0x63, 0x61, 0x63, 0x68, 0x65, 0x5f,
     0x6b, 0x65, 0x79, 0x5f, 0x66, 0x6f, 0x72, 0x6d, 0x61, 0x74, 0x01, 0x81, 0xab, 0x77, 0x69, 0x72, 0x65, 0x2d, 0x62, 0x75, 0x63,
-    0x6b, 0x65, 0x74, 0x8b, 0xa8, 0x63, 0x68, 0x69, 0x6c, 0x64, 0x72, 0x65, 0x6e, 0x90, 0xa4, 0x73, 0x69, 0x7a, 0x65, 0xcd, 0x10,
+    0x6b, 0x65, 0x74, 0x8c, 0xa8, 0x63, 0x68, 0x69, 0x6c, 0x64, 0x72, 0x65, 0x6e, 0x90, 0xa4, 0x73, 0x69, 0x7a, 0x65, 0xcd, 0x10,
     0x00, 0xa7, 0x6f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x73, 0x03, 0xa8, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x05, 0xae,
     0x64, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x5f, 0x6d, 0x61, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x01, 0xa9, 0x6f, 0x62, 0x6a, 0x5f, 0x73,
     0x69, 0x7a, 0x65, 0x73, 0x9b, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xac, 0x6f, 0x62, 0x6a, 0x5f,
@@ -934,7 +1109,8 @@ const USAGE_CACHE_WIRE_FIXTURE: &[u8] = &[
     0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x73, 0xc0, 0xa9, 0x63, 0x6f, 0x6d, 0x70, 0x61, 0x63,
     0x74, 0x65, 0x64, 0xc3, 0xae, 0x66, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x5f, 0x6f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x73, 0x02, 0xae,
     0x61, 0x6c, 0x6c, 0x5f, 0x74, 0x69, 0x65, 0x72, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x73, 0x91, 0x81, 0xa4, 0x57, 0x41, 0x52, 0x4d,
-    0x93, 0xcd, 0x08, 0x00, 0x02, 0x01,
+    0x93, 0xcd, 0x08, 0x00, 0x02, 0x01, 0xb2, 0x75, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x5f, 0x74, 0x69, 0x65, 0x72, 0x5f, 0x73,
+    0x74, 0x61, 0x74, 0x73, 0xc0,
 ];
 
 #[test]

crates/scanner/src/lib.rs

@@ -94,6 +94,45 @@ static SCANNER_RUNTIME_INSTANCES: AtomicU64 = AtomicU64::new(0);
 static SCANNER_FOREGROUND_READ_ACTIVITY: AtomicU64 = AtomicU64::new(0);
 static SCANNER_FOREGROUND_STREAM_READS: AtomicU64 = AtomicU64::new(0);
 
+/// Immutable tier registry captured at the beginning of a folder scan.
+/// Generation makes it possible to prove that a result was classified against
+/// one registry even when the process-wide TTL cache refreshes concurrently.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct TierRegistrySnapshot {
+    pub(crate) generation: u64,
+    pub(crate) names: Arc<[String]>,
+    /// True when the last refresh attempt failed and `names` is therefore a
+    /// retained last-good snapshot rather than a newly read registry.
+    pub(crate) refresh_failed: bool,
+}
+
+impl TierRegistrySnapshot {
+    /// Apply a refresh only when the registry read succeeds. A failed refresh
+    /// retains the prior generation, preventing a transient config failure
+    /// from classifying the remainder of a scan against an empty registry.
+    pub(crate) fn refreshed(&self, names: Result<Arc<[String]>, ()>) -> Self {
+        match names {
+            Ok(names) => Self {
+                generation: self.generation.saturating_add(1),
+                names,
+                refresh_failed: false,
+            },
+            Err(()) => Self {
+                refresh_failed: true,
+                ..self.clone()
+            },
+        }
+    }
+
+    pub(crate) fn initial(names: Arc<[String]>) -> Self {
+        Self {
+            generation: 1,
+            names,
+            refresh_failed: false,
+        }
+    }
+}
+
 pub fn current_scanner_activity() -> u64 {
     SCANNER_ACTIVE_WORK_UNITS.load(Ordering::Relaxed)
 }
@@ -383,24 +422,48 @@ const TIER_NAME_CACHE_TTL: Duration = Duration::from_secs(30);
 /// `TIER_NAME_CACHE_TTL` later; a removed tier can leave an all-zero
 /// `TierStats` seed behind for one cache generation, which merges harmlessly
 /// by key in per-object accounting and disappears on the next refresh.
-static TIER_NAME_CACHE: RwLock<Option<(Instant, Arc<[String]>)>> = RwLock::new(None);
+static TIER_NAME_CACHE: RwLock<Option<(Instant, TierRegistrySnapshot)>> = RwLock::new(None);
+static TIER_REGISTRY_GENERATION: AtomicU64 = AtomicU64::new(0);
 
-/// Tier names currently registered in the tier configuration, cached for
-/// `TIER_NAME_CACHE_TTL`.
-pub(crate) async fn runtime_tier_names() -> Arc<[String]> {
+/// Return one immutable registry snapshot for a scanner unit of work.
+pub(crate) async fn runtime_tier_registry() -> TierRegistrySnapshot {
     {
         let cached = TIER_NAME_CACHE.read().unwrap_or_else(|err| err.into_inner()).clone();
-        if let Some((refreshed_at, names)) = cached
+        if let Some((refreshed_at, snapshot)) = cached
             && refreshed_at.elapsed() < TIER_NAME_CACHE_TTL
         {
-            return names;
+            return snapshot;
         }
     }
 
     let tiers = ecstore_get_global_tier_config_mgr().read().await.list_tiers();
     let names: Arc<[String]> = tiers.iter().map(|tier| tier.name.clone()).collect::<Vec<_>>().into();
-    *TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = Some((Instant::now(), Arc::clone(&names)));
-    names
+    let previous = TIER_NAME_CACHE
+        .read()
+        .unwrap_or_else(|err| err.into_inner())
+        .as_ref()
+        .map(|(_, snapshot)| snapshot.clone());
+    let snapshot = previous
+        .as_ref()
+        .map(|snapshot| {
+            let refreshed = snapshot.refreshed(Ok(Arc::clone(&names)));
+            TierRegistrySnapshot {
+                generation: TIER_REGISTRY_GENERATION.fetch_add(1, Ordering::Relaxed).saturating_add(1),
+                ..refreshed
+            }
+        })
+        .unwrap_or_else(|| TierRegistrySnapshot {
+            generation: TIER_REGISTRY_GENERATION.fetch_add(1, Ordering::Relaxed).saturating_add(1),
+            ..TierRegistrySnapshot::initial(names)
+        });
+    *TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = Some((Instant::now(), snapshot.clone()));
+    snapshot
+}
+
+/// Tier names currently registered in the tier configuration, cached for
+/// `TIER_NAME_CACHE_TTL`.
+pub(crate) async fn runtime_tier_names() -> Arc<[String]> {
+    runtime_tier_registry().await.names
 }
 
 /// Test-only cache reset; the production cache has no invalidation hook

crates/scanner/src/runtime_config.rs

@@ -125,7 +125,10 @@ impl Default for ScannerRuntimeConfig {
             cycle_interval_source: ScannerRuntimeConfigSource::Default,
             bitrot_cycle: Some(Duration::from_secs(DEFAULT_HEAL_BITROT_CYCLE_SECS)),
             bitrot_cycle_source: ScannerRuntimeConfigSource::Default,
-            cycle_budget: ScannerCycleBudgetConfig::default(),
+            cycle_budget: ScannerCycleBudgetConfig {
+                max_duration: Some(Duration::from_secs(DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)),
+                ..Default::default()
+            },
             cycle_max_duration_source: ScannerRuntimeConfigSource::Default,
             cycle_max_objects_source: ScannerRuntimeConfigSource::Default,
             cycle_max_directories_source: ScannerRuntimeConfigSource::Default,
@@ -374,7 +377,10 @@ fn validate_persisted_scanner_runtime_config(config: &ServerConfig) -> Result<()
     }
     validate_optional_config_u64(scanner_kvs, SCANNER_START_DELAY, "")?;
     validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE, "")?;
-    validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)?;
+    if let Some(value) = config_value(scanner_kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS) {
+        let secs = parse_config_u64(SCANNER_CYCLE_MAX_DURATION, value)?;
+        cycle_duration_from_secs(SCANNER_CYCLE_MAX_DURATION, secs)?;
+    }
     validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS)?;
     validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES)?;
     if let Some(value) = config_value(heal_kvs, HEAL_BITROT_CYCLE, DEFAULT_HEAL_BITROT_CYCLE_SECS) {
@@ -436,19 +442,46 @@ fn lookup_max_wait(
     Ok((speed.max_sleep(), speed_source))
 }
 
-fn lookup_optional_seconds(
-    kvs: Option<&KVS>,
-    key: &'static str,
-    env_key: &'static str,
-    default: u64,
-) -> Result<(Option<Duration>, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
-    if let Some(secs) = rustfs_utils::get_env_opt_u64(env_key) {
-        return Ok((Some(Duration::from_secs(secs)), ScannerRuntimeConfigSource::Env));
+fn lookup_cycle_duration(kvs: Option<&KVS>) -> Result<(Option<Duration>, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
+    match rustfs_utils::get_env_parse_outcome::<u64>(ENV_SCANNER_CYCLE_MAX_DURATION_SECS) {
+        rustfs_utils::EnvParseOutcome::Parsed(secs) => {
+            return cycle_duration_from_secs(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, secs)
+                .map(|duration| (duration, ScannerRuntimeConfigSource::Env));
+        }
+        rustfs_utils::EnvParseOutcome::Invalid => {
+            // Do not include the raw environment value in the typed error:
+            // deployments occasionally put sensitive material in inherited
+            // environment snapshots. The key still identifies the control.
+            return Err(invalid_value(
+                ENV_SCANNER_CYCLE_MAX_DURATION_SECS,
+                "<invalid>",
+                "expected unsigned integer seconds",
+            ));
+        }
+        rustfs_utils::EnvParseOutcome::Absent => {}
     }
-    if let Some(value) = config_value(kvs, key, default) {
-        return parse_config_u64(key, value).map(|secs| (Some(Duration::from_secs(secs)), ScannerRuntimeConfigSource::Config));
+
+    if let Some(value) = config_value(kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS) {
+        let secs = parse_config_u64(SCANNER_CYCLE_MAX_DURATION, value)?;
+        return cycle_duration_from_secs(SCANNER_CYCLE_MAX_DURATION, secs)
+            .map(|duration| (duration, ScannerRuntimeConfigSource::Config));
     }
-    Ok((None, ScannerRuntimeConfigSource::Default))
+
+    Ok((
+        Some(Duration::from_secs(DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)),
+        ScannerRuntimeConfigSource::Default,
+    ))
+}
+
+fn cycle_duration_from_secs(key: &'static str, secs: u64) -> Result<Option<Duration>, ScannerRuntimeConfigError> {
+    if secs == 0 {
+        return Ok(None);
+    }
+    let duration = Duration::from_secs(secs);
+    if std::time::Instant::now().checked_add(duration).is_none() {
+        return Err(invalid_value(key, "<overflow>", "duration exceeds the timer range"));
+    }
+    Ok(Some(duration))
 }
 
 fn lookup_start_delay(kvs: Option<&KVS>) -> Result<(Option<Duration>, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
@@ -553,12 +586,7 @@ pub(crate) fn lookup_scanner_runtime_config(
         (speed.cycle_interval(), speed_source)
     };
 
-    let (cycle_max_duration, cycle_max_duration_source) = lookup_optional_seconds(
-        scanner_kvs,
-        SCANNER_CYCLE_MAX_DURATION,
-        ENV_SCANNER_CYCLE_MAX_DURATION_SECS,
-        DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS,
-    )?;
+    let (cycle_max_duration, cycle_max_duration_source) = lookup_cycle_duration(scanner_kvs)?;
     let (cycle_max_objects, cycle_max_objects_source) = lookup_count_budget(
         scanner_kvs,
         SCANNER_CYCLE_MAX_OBJECTS,
@@ -863,10 +891,10 @@ mod tests {
     use rustfs_config::server_config::{Config as ServerConfig, KVS};
     use rustfs_config::{
         DEFAULT_DELIMITER, DEFAULT_HEAL_BITROT_CYCLE_SECS, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS,
-        ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY, ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED,
-        HEAL_BITROT_CYCLE, HEAL_SUB_SYS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE,
-        SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE,
-        SCANNER_SPEED, SCANNER_SUB_SYS, ScannerSpeed,
+        ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY,
+        ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED, HEAL_BITROT_CYCLE, HEAL_SUB_SYS, SCANNER_BITROT_CYCLE,
+        SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION,
+        SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE, SCANNER_SPEED, SCANNER_SUB_SYS, ScannerSpeed,
     };
     use std::collections::HashMap;
     use std::time::Duration;
@@ -941,6 +969,50 @@ mod tests {
         });
     }
 
+    #[test]
+    fn scanner_unset_budget_uses_safe_default_but_explicit_zero_is_unbounded() {
+        let config = server_config_with_scanner(&[]);
+        with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || {
+            let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
+            assert_eq!(resolved.cycle_budget.max_duration, Some(Duration::from_secs(1800)));
+            assert_eq!(resolved.cycle_max_duration_source, ScannerRuntimeConfigSource::Default);
+        });
+
+        let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "0")]);
+        with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || {
+            let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
+            assert_eq!(resolved.cycle_budget.max_duration, None);
+            assert_eq!(resolved.cycle_max_duration_source, ScannerRuntimeConfigSource::Config);
+        });
+    }
+
+    #[test]
+    fn cycle_budget_invalid_or_overflow_config_is_rejected() {
+        with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("invalid"), || {
+            let error = lookup_scanner_runtime_config(None).expect_err("invalid duration env must be rejected");
+            assert!(error.to_string().contains(ENV_SCANNER_CYCLE_MAX_DURATION_SECS));
+            assert!(error.to_string().contains("<invalid>"));
+            assert!(!error.to_string().contains(": invalid ("));
+        });
+        with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("18446744073709551616"), || {
+            assert!(lookup_scanner_runtime_config(None).is_err());
+        });
+        with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("18446744073709551615"), || {
+            assert!(lookup_scanner_runtime_config(None).is_err());
+        });
+        let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "not-a-duration")]);
+        assert!(lookup_scanner_runtime_config(Some(&config)).is_err());
+    }
+
+    #[test]
+    fn scanner_runtime_config_validation_rejects_overflow_persisted_duration() {
+        let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "18446744073709551615")]);
+
+        let error = validate_scanner_runtime_config(&config)
+            .expect_err("persisted duration that exceeds the timer range must be rejected");
+        assert!(error.to_string().contains(SCANNER_CYCLE_MAX_DURATION));
+    }
+
     #[test]
     fn scanner_runtime_config_normalizes_persisted_default_speed() {
         let config = server_config_with_scanner(&[(SCANNER_SPEED, "default")]);

crates/scanner/src/scanner.rs

@@ -52,6 +52,7 @@ use rustfs_config::{
 };
 use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS};
 use rustfs_data_usage::observed_data_usage_is_newer;
+use rustfs_lock::NamespaceLockGuard;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest as _, Sha256};
 use tokio::sync::{Notify, mpsc};
@@ -1037,20 +1038,116 @@ fn data_usage_persist_timeout() -> Duration {
     DataUsageCache::persistence_timeout()
 }
 
+#[cfg(not(test))]
+const SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT: Duration = Duration::from_secs(30);
+#[cfg(test)]
+const SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT: Duration = Duration::from_millis(50);
+
+async fn fence_scanner_epoch_after_cycle_timeout<Store, LockLost>(
+    ctx: &CancellationToken,
+    storeapi: Arc<Store>,
+    cycle_info: &mut CurrentCycle,
+    cycle_revision: &mut DataUsageCacheRevision,
+    leader_epoch: &mut u64,
+    lock_lost: LockLost,
+) -> bool
+where
+    Store: ScannerObjectIO,
+    LockLost: Future<Output = ()>,
+{
+    let fence_ctx = ctx.child_token();
+    let claim = claim_scanner_leadership(&fence_ctx, storeapi, cycle_info, cycle_revision, leader_epoch);
+    tokio::pin!(claim);
+    tokio::pin!(lock_lost);
+    tokio::select! {
+        biased;
+        _ = &mut lock_lost => {
+            fence_ctx.cancel();
+            false
+        }
+        result = tokio::time::timeout(SCANNER_CYCLE_EPOCH_FENCE_TIMEOUT, &mut claim) => {
+            result.unwrap_or(false) && !fence_ctx.is_cancelled()
+        }
+    }
+}
+
+struct ScannerCycleDeadlineState<'a> {
+    cycle_info: &'a mut CurrentCycle,
+    cycle_revision: &'a mut DataUsageCacheRevision,
+    leader_epoch: &'a mut u64,
+    cycle_budget: &'a ScannerCycleBudget,
+}
+
+fn cycle_timeout_requires_recovery(worker_stopped: bool, cycle_state_persisted: bool, generation_fenced: bool) -> bool {
+    !worker_stopped || !cycle_state_persisted || !generation_fenced
+}
+
+async fn handle_scanner_cycle_deadline<Store>(
+    ctx: &CancellationToken,
+    storeapi: Arc<Store>,
+    state: ScannerCycleDeadlineState<'_>,
+    worker_stopped: bool,
+    guard: &mut NamespaceLockGuard,
+) where
+    Store: ScannerObjectIO,
+{
+    let fenced = fence_scanner_epoch_after_cycle_timeout(
+        ctx,
+        storeapi,
+        state.cycle_info,
+        state.cycle_revision,
+        state.leader_epoch,
+        guard.lock_lost_notified(),
+    )
+    .await;
+    let cycle_state_persisted = state.cycle_budget.cycle_state_persisted();
+    let recovery_required = cycle_timeout_requires_recovery(worker_stopped, cycle_state_persisted, fenced);
+    warn!(
+        target: "rustfs::scanner",
+        event = EVENT_SCANNER_CYCLE_STATE,
+        component = LOG_COMPONENT_SCANNER,
+        subsystem = LOG_SUBSYSTEM_RUNTIME,
+        state = "cycle_timeout",
+        worker_stopped,
+        cycle_state_persisted,
+        generation_fenced = fenced,
+        recovery_required,
+        "Scanner cycle deadline expired; durable cursor/generation fencing completed when possible"
+    );
+    global_metrics().record_scanner_cycle_timeout(recovery_required, state.cycle_budget.progress_age());
+    // Stop renewing before releasing the lease. A new leader can then claim the
+    // higher persisted generation instead of inheriting the expired worker.
+    guard.release();
+    global_metrics().set_cycle(None).await;
+}
+
 async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle, cycle_metrics_guard: &mut ScannerCycleMetricsGuard) {
     cycle_info.current = 0;
     global_metrics().clear_current_scan_mode();
     cycle_metrics_guard.finish(cycle_info.clone()).await;
 }
 
-#[instrument(skip_all)]
-#[hotpath::measure]
+#[cfg(test)]
 async fn run_data_scanner_cycle(
     ctx: &CancellationToken,
     storeapi: &Arc<ECStore>,
     cycle_info: &mut CurrentCycle,
     cycle_revision: &mut DataUsageCacheRevision,
     leader_epoch: u64,
+) -> ScannerCycleOutcome {
+    let cycle_budget = ScannerCycleBudget::new(ctx, scanner_cycle_budget_config());
+    run_data_scanner_cycle_with_budget(ctx, storeapi, cycle_info, cycle_revision, leader_epoch, cycle_budget).await
+}
+
+#[instrument(skip_all)]
+#[hotpath::measure]
+async fn run_data_scanner_cycle_with_budget(
+    ctx: &CancellationToken,
+    storeapi: &Arc<ECStore>,
+    cycle_info: &mut CurrentCycle,
+    cycle_revision: &mut DataUsageCacheRevision,
+    leader_epoch: u64,
+    cycle_budget: Arc<ScannerCycleBudget>,
 ) -> ScannerCycleOutcome {
     let _activity_guard = ScannerActivityGuard::new();
     if let Err(err) = refresh_scanner_runtime_config_from_global() {
@@ -1066,7 +1163,11 @@ async fn run_data_scanner_cycle(
     }
     let configured_cycle_interval = scanner_cycle_interval();
     let configured_bitrot_cycle = scanner_bitrot_cycle();
-    let cycle_budget_config = scanner_cycle_budget_config();
+    let cycle_budget_config = ScannerCycleBudgetConfig {
+        max_duration: cycle_budget.max_duration(),
+        max_objects: cycle_budget.max_objects(),
+        max_directories: cycle_budget.max_directories(),
+    };
     let usage_persist_timeout = data_usage_persist_timeout();
     global_metrics().record_scanner_cycle_config(
         configured_cycle_interval,
@@ -1137,7 +1238,6 @@ async fn run_data_scanner_cycle(
     let (sender, receiver) = mpsc::channel::<DataUsageInfo>(1);
 
     let done_cycle = Metrics::time(Metric::ScanCycle);
-    let cycle_budget = ScannerCycleBudget::new(ctx, cycle_budget_config);
     let scan_result = storeapi
         .clone()
         .nsscanner_with_status(
@@ -1277,7 +1377,7 @@ async fn run_data_scanner_cycle(
                 "Scanner cycle is recovering to a newer durable cache generation"
             );
             emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None);
-            return if persist_required_scanner_cycle_floor(
+            let persisted = persist_required_scanner_cycle_floor(
                 ctx,
                 storeapi.clone(),
                 cycle_info,
@@ -1286,8 +1386,9 @@ async fn run_data_scanner_cycle(
                 required_cycle,
                 &mut cycle_metrics_guard,
             )
-            .await
-            {
+            .await;
+            return if persisted {
+                cycle_budget.mark_cycle_state_persisted();
                 ScannerCycleOutcome::Partial
             } else {
                 ScannerCycleOutcome::Failed
@@ -1345,7 +1446,7 @@ async fn run_data_scanner_cycle(
             scan_cycle_partial_reason(budget_reason),
             scan_cycle_partial_source(budget_reason),
         );
-        return if finalize_partial_scan_cycle(
+        let persisted = finalize_partial_scan_cycle(
             ctx,
             storeapi.clone(),
             cycle_info,
@@ -1353,8 +1454,9 @@ async fn run_data_scanner_cycle(
             leader_epoch,
             &mut cycle_metrics_guard,
         )
-        .await
-        {
+        .await;
+        return if persisted {
+            cycle_budget.mark_cycle_state_persisted();
             ScannerCycleOutcome::Partial
         } else {
             ScannerCycleOutcome::Failed
@@ -1429,7 +1531,7 @@ async fn run_data_scanner_cycle(
                 );
             }
             emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None);
-            return if finalize_partial_scan_cycle(
+            let persisted = finalize_partial_scan_cycle(
                 ctx,
                 storeapi.clone(),
                 cycle_info,
@@ -1437,8 +1539,9 @@ async fn run_data_scanner_cycle(
                 leader_epoch,
                 &mut cycle_metrics_guard,
             )
-            .await
-            {
+            .await;
+            return if persisted {
+                cycle_budget.mark_cycle_state_persisted();
                 ScannerCycleOutcome::Partial
             } else {
                 ScannerCycleOutcome::Failed
@@ -1479,6 +1582,7 @@ async fn run_data_scanner_cycle(
             )
             .await
             {
+                cycle_budget.mark_cycle_state_persisted();
                 emit_scan_cycle_superseded(cycle_start.elapsed());
                 return ScannerCycleOutcome::Superseded;
             }
@@ -1511,6 +1615,7 @@ async fn run_data_scanner_cycle(
         emit_scan_cycle_complete(false, cycle_start.elapsed());
         return ScannerCycleOutcome::Failed;
     }
+    cycle_budget.mark_cycle_state_persisted();
 
     done_cycle();
     emit_scan_cycle_complete(true, cycle_start.elapsed());
@@ -1575,7 +1680,7 @@ async fn run_data_scanner_with_maintenance_state(
 ) -> Result<(), ScannerError> {
     reset_scanner_cycle_schedule();
     // Acquire leader lock (write lock) to ensure only one scanner runs
-    let guard = match storeapi.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock").await {
+    let mut guard = match storeapi.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock").await {
         Ok(ns_lock) => match ns_lock.get_write_lock_quiet(get_lock_acquire_timeout()).await {
             Ok(guard) => {
                 record_scanner_leader_lock_state("acquired");
@@ -1740,13 +1845,49 @@ async fn run_data_scanner_with_maintenance_state(
             return Ok(());
         }
         let cycle_ctx = ctx.child_token();
-        let initial_outcome = await_scanner_cycle_with_lock_fence(
+        let cycle_budget = ScannerCycleBudget::new_with_runtime_progress_tracking(&cycle_ctx, scanner_cycle_budget_config());
+        let initial_outcome = match await_scanner_cycle_with_budget_fence(
             &cycle_ctx,
-            run_data_scanner_cycle(&cycle_ctx, &storeapi, &mut cycle_info, &mut cycle_revision, leader_epoch),
+            &cycle_budget,
+            run_data_scanner_cycle_with_budget(
+                &cycle_ctx,
+                &storeapi,
+                &mut cycle_info,
+                &mut cycle_revision,
+                leader_epoch,
+                cycle_budget.clone(),
+            ),
             guard.lock_lost_notified(),
         )
         .await
-        .unwrap_or(ScannerCycleOutcome::Failed);
+        {
+            ScannerCycleWaitOutcome::Completed(outcome) => outcome,
+            ScannerCycleWaitOutcome::LockLost => {
+                record_scanner_leader_lock_lost("Scanner leader lock lost during the initial cycle").await;
+                global_metrics().set_cycle(None).await;
+                return Ok(());
+            }
+            ScannerCycleWaitOutcome::Cancelled => {
+                global_metrics().set_cycle(None).await;
+                return Ok(());
+            }
+            ScannerCycleWaitOutcome::Deadline { worker_stopped } => {
+                handle_scanner_cycle_deadline(
+                    &ctx,
+                    storeapi.clone(),
+                    ScannerCycleDeadlineState {
+                        cycle_info: &mut cycle_info,
+                        cycle_revision: &mut cycle_revision,
+                        leader_epoch: &mut leader_epoch,
+                        cycle_budget: &cycle_budget,
+                    },
+                    worker_stopped,
+                    &mut guard,
+                )
+                .await;
+                return Ok(());
+            }
+        };
         superseded_backoff.record_retryable_cycle(initial_outcome == ScannerCycleOutcome::Superseded);
         deferred_backoff.record_retryable_cycle(matches!(initial_outcome, ScannerCycleOutcome::Deferred(_)));
         dirty_usage_generation_seen = dirty_generation_before_cycle;
@@ -1952,13 +2093,49 @@ async fn run_data_scanner_with_maintenance_state(
         }
         let dirty_generation_before_cycle = dirty_usage_generation();
         let cycle_ctx = ctx.child_token();
-        let outcome = await_scanner_cycle_with_lock_fence(
+        let cycle_budget = ScannerCycleBudget::new_with_runtime_progress_tracking(&cycle_ctx, scanner_cycle_budget_config());
+        let outcome = match await_scanner_cycle_with_budget_fence(
             &cycle_ctx,
-            run_data_scanner_cycle(&cycle_ctx, &storeapi, &mut cycle_info, &mut cycle_revision, leader_epoch),
+            &cycle_budget,
+            run_data_scanner_cycle_with_budget(
+                &cycle_ctx,
+                &storeapi,
+                &mut cycle_info,
+                &mut cycle_revision,
+                leader_epoch,
+                cycle_budget.clone(),
+            ),
             guard.lock_lost_notified(),
         )
         .await
-        .unwrap_or(ScannerCycleOutcome::Failed);
+        {
+            ScannerCycleWaitOutcome::Completed(outcome) => outcome,
+            ScannerCycleWaitOutcome::LockLost => {
+                record_scanner_leader_lock_lost("Scanner leader lock lost during a scanner cycle").await;
+                global_metrics().set_cycle(None).await;
+                return Ok(());
+            }
+            ScannerCycleWaitOutcome::Cancelled => {
+                global_metrics().set_cycle(None).await;
+                return Ok(());
+            }
+            ScannerCycleWaitOutcome::Deadline { worker_stopped } => {
+                handle_scanner_cycle_deadline(
+                    &ctx,
+                    storeapi.clone(),
+                    ScannerCycleDeadlineState {
+                        cycle_info: &mut cycle_info,
+                        cycle_revision: &mut cycle_revision,
+                        leader_epoch: &mut leader_epoch,
+                        cycle_budget: &cycle_budget,
+                    },
+                    worker_stopped,
+                    &mut guard,
+                )
+                .await;
+                return Ok(());
+            }
+        };
         superseded_backoff.record_retryable_cycle(outcome == ScannerCycleOutcome::Superseded);
         deferred_backoff.record_retryable_cycle(matches!(outcome, ScannerCycleOutcome::Deferred(_)));
         dirty_usage_generation_seen = dirty_generation_before_cycle;

crates/scanner/src/scanner/cycle_state.rs

@@ -1581,3 +1581,63 @@ where
         output = &mut cycle => Some(output),
     }
 }
+
+#[derive(Debug, PartialEq, Eq)]
+pub(super) enum ScannerCycleWaitOutcome<T> {
+    Completed(T),
+    LockLost,
+    Cancelled,
+    Deadline { worker_stopped: bool },
+}
+
+pub(super) async fn await_scanner_cycle_with_budget_fence<Cycle, LockLost>(
+    cycle_ctx: &CancellationToken,
+    budget: &ScannerCycleBudget,
+    cycle: Cycle,
+    lock_lost: LockLost,
+) -> ScannerCycleWaitOutcome<Cycle::Output>
+where
+    Cycle: Future,
+    LockLost: Future<Output = ()>,
+{
+    tokio::pin!(cycle);
+    tokio::pin!(lock_lost);
+    let deadline = async {
+        if let Some(deadline) = budget.deadline() {
+            tokio::time::sleep_until(deadline).await;
+        } else {
+            std::future::pending::<()>().await;
+        }
+    };
+    tokio::pin!(deadline);
+    tokio::select! {
+        biased;
+        _ = &mut lock_lost => {
+            cycle_ctx.cancel();
+            let _ = tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, &mut cycle).await;
+            ScannerCycleWaitOutcome::LockLost
+        }
+        _ = &mut deadline => {
+            budget.cancel_for_runtime();
+            // Let the budget cancellation reach the scanner first so it can
+            // persist a partial cursor. Only an uncooperative worker gets the
+            // parent cancellation, and it is dropped after the bounded window;
+            // the caller fences its epoch next.
+            let worker_stopped = if tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, &mut cycle)
+                .await
+                .is_ok()
+            {
+                true
+            } else {
+                cycle_ctx.cancel();
+                false
+            };
+            ScannerCycleWaitOutcome::Deadline { worker_stopped }
+        }
+        _ = cycle_ctx.cancelled() => {
+            let _ = tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, &mut cycle).await;
+            ScannerCycleWaitOutcome::Cancelled
+        }
+        output = &mut cycle => ScannerCycleWaitOutcome::Completed(output),
+    }
+}

crates/scanner/src/scanner/tests.rs

@@ -26,6 +26,7 @@ use std::task::Poll;
 use temp_env::{with_var, with_var_unset};
 use tokio::io::AsyncReadExt;
 use tokio::sync::Mutex;
+use tokio::time::{Duration, advance};
 
 const TEST_DEFAULT_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60;
 
@@ -118,6 +119,178 @@ async fn scanner_cycle_lock_fence_bounds_uncooperative_shutdown() {
     assert!(cycle_ctx.is_cancelled());
 }
 
+#[tokio::test(start_paused = true)]
+async fn cycle_budget_fences_late_writer_after_timeout() {
+    let cycle_ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(
+        &cycle_ctx,
+        ScannerCycleBudgetConfig {
+            max_duration: Some(Duration::from_secs(5)),
+            ..Default::default()
+        },
+    );
+    let outcome = {
+        let cycle = std::future::pending::<()>();
+        let lock_lost = std::future::pending::<()>();
+        let waiter = await_scanner_cycle_with_budget_fence(&cycle_ctx, &budget, cycle, lock_lost);
+        tokio::pin!(waiter);
+        tokio::task::yield_now().await;
+        advance(Duration::from_secs(5)).await;
+        tokio::task::yield_now().await;
+        advance(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT).await;
+        waiter.await
+    };
+    assert_eq!(outcome, ScannerCycleWaitOutcome::Deadline { worker_stopped: false });
+    assert!(cycle_ctx.is_cancelled());
+    assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Runtime));
+
+    // A newer leadership epoch is the durable fence that rejects a late
+    // writer after the timed-out future has been dropped.
+    let store = Arc::new(MemoryConfigStore::default());
+    let mut revision = DataUsageCacheRevision::Missing;
+    let mut cycle = CurrentCycle {
+        current: 0,
+        next: 12,
+        ..Default::default()
+    };
+    let persist_ctx = CancellationToken::new();
+    assert!(persist_scanner_cycle_state(&persist_ctx, store.clone(), &mut cycle, &mut revision, 1).await);
+    let newer = encode_scanner_cycle_state(&cycle, 2).expect("new epoch fence should encode");
+    let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
+    store.interleaving_puts.lock().await.insert(key, (2, newer));
+    let mut late_cycle = CurrentCycle { next: 13, ..cycle };
+    assert!(!persist_scanner_cycle_state(&persist_ctx, store, &mut late_cycle, &mut revision, 1).await);
+}
+
+#[tokio::test(start_paused = true)]
+async fn cycle_budget_parent_cancellation_is_not_reported_as_timeout() {
+    let cycle_ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(
+        &cycle_ctx,
+        ScannerCycleBudgetConfig {
+            max_duration: Some(Duration::from_secs(5)),
+            ..Default::default()
+        },
+    );
+    let waiter = await_scanner_cycle_with_budget_fence(&cycle_ctx, &budget, std::future::pending::<()>(), std::future::pending());
+    tokio::pin!(waiter);
+    tokio::task::yield_now().await;
+    cycle_ctx.cancel();
+    tokio::task::yield_now().await;
+    advance(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT).await;
+    assert_eq!(waiter.await, ScannerCycleWaitOutcome::Cancelled);
+}
+
+#[tokio::test(start_paused = true)]
+async fn cycle_budget_deadline_wins_same_tick_as_parent_cancellation() {
+    let cycle_ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(
+        &cycle_ctx,
+        ScannerCycleBudgetConfig {
+            max_duration: Some(Duration::from_secs(5)),
+            ..Default::default()
+        },
+    );
+    let waiter = await_scanner_cycle_with_budget_fence(&cycle_ctx, &budget, std::future::pending::<()>(), std::future::pending());
+    tokio::pin!(waiter);
+    tokio::task::yield_now().await;
+    advance(Duration::from_secs(5)).await;
+    cycle_ctx.cancel();
+    tokio::task::yield_now().await;
+    advance(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT).await;
+
+    assert_eq!(waiter.await, ScannerCycleWaitOutcome::Deadline { worker_stopped: false });
+    assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Runtime));
+}
+
+#[tokio::test]
+async fn cycle_budget_persist_cursor_failure_is_recovery_required() {
+    let store = Arc::new(MemoryConfigStore::default());
+    let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
+    store.fail_put_number.lock().await.insert(key, 1);
+
+    let ctx = CancellationToken::new();
+    let mut revision = DataUsageCacheRevision::Missing;
+    let mut cycle = CurrentCycle {
+        current: 12,
+        next: 12,
+        ..Default::default()
+    };
+    let mut leader_epoch = 1;
+    let fenced = fence_scanner_epoch_after_cycle_timeout(
+        &ctx,
+        store,
+        &mut cycle,
+        &mut revision,
+        &mut leader_epoch,
+        std::future::pending(),
+    )
+    .await;
+    assert!(!fenced, "a failed cursor/generation write must require recovery");
+    let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+    assert!(cycle_timeout_requires_recovery(true, budget.cycle_state_persisted(), fenced));
+
+    let metrics = Metrics::new();
+    metrics.record_scanner_cycle_timeout(!fenced, Duration::from_secs(17));
+    let report = metrics.report().await;
+    assert_eq!(report.cycle_timeout_total, 1);
+    assert_eq!(report.cycle_recovery_required_total, 1);
+    assert_eq!(report.cycle_last_progress_age, 17);
+    assert!(report.leader_lease_without_progress);
+}
+
+#[tokio::test]
+async fn cycle_budget_deadline_handler_fences_and_releases_guard() {
+    let (_temp_dir, store) = setup_scanner_cycle_store().await;
+    let lock = store
+        .new_ns_lock(RUSTFS_META_BUCKET, "leader.lock")
+        .await
+        .expect("scanner leader lock should be created");
+    let mut guard = lock
+        .get_write_lock(Duration::from_secs(1))
+        .await
+        .expect("scanner leader lock should be acquired");
+
+    let ctx = CancellationToken::new();
+    let mut cycle_info = CurrentCycle {
+        current: 12,
+        next: 12,
+        ..Default::default()
+    };
+    let mut cycle_revision = DataUsageCacheRevision::Missing;
+    let mut leader_epoch = 1;
+    let budget = ScannerCycleBudget::new(
+        &ctx,
+        ScannerCycleBudgetConfig {
+            max_duration: Some(Duration::from_secs(60)),
+            ..Default::default()
+        },
+    );
+    budget.mark_cycle_state_persisted();
+
+    handle_scanner_cycle_deadline(
+        &ctx,
+        store.clone(),
+        ScannerCycleDeadlineState {
+            cycle_info: &mut cycle_info,
+            cycle_revision: &mut cycle_revision,
+            leader_epoch: &mut leader_epoch,
+            cycle_budget: &budget,
+        },
+        true,
+        &mut guard,
+    )
+    .await;
+
+    assert!(guard.is_released());
+    let persisted = read_config(store, &DATA_USAGE_BLOOM_NAME_PATH)
+        .await
+        .expect("deadline handler should persist a fenced cursor");
+    let (_, persisted_epoch) = decode_scanner_cycle_state(&persisted).expect("fenced cursor should decode");
+    assert_eq!(persisted_epoch, 2);
+    global_metrics().set_cycle(None).await;
+}
+
 #[tokio::test]
 async fn scanner_cycle_recovery_wake_survives_wait_registration_race() {
     notify_scanner_cycle_recovery_wake();
@@ -428,13 +601,6 @@ fn test_scanner_cycle_max_duration_uses_env() {
     });
 }
 
-#[test]
-fn test_scanner_cycle_max_duration_default_is_disabled() {
-    with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || {
-        assert_eq!(scanner_cycle_max_duration(), None);
-    });
-}
-
 #[tokio::test]
 async fn test_scanner_cycle_budget_cancels_after_duration() {
     let parent = CancellationToken::new();
@@ -2242,7 +2408,7 @@ async fn test_leadership_claim_usage_fence_rejects_old_inflight_writer() {
 }
 
 #[tokio::test]
-async fn test_successful_old_epoch_commit_is_fenced_after_cancellation() {
+async fn cycle_budget_lease_takeover_rejects_old_generation() {
     let store = Arc::new(MemoryConfigStore::default());
     let ctx = CancellationToken::new();
     let mut revision = DataUsageCacheRevision::Missing;
@@ -2287,12 +2453,17 @@ async fn test_successful_old_epoch_commit_is_fenced_after_cancellation() {
         .await
     );
 
-    let state = read_config(store, &DATA_USAGE_BLOOM_NAME_PATH)
+    let state = read_config(store.clone(), &DATA_USAGE_BLOOM_NAME_PATH)
         .await
         .expect("replacement leadership claim should persist");
     let (claimed_cycle, claimed_epoch) = decode_scanner_cycle_state(&state).expect("replacement cycle state should decode");
     assert_eq!(claimed_cycle.next, 14);
     assert_eq!(claimed_epoch, 2);
+
+    let mut stale_cycle = CurrentCycle { next: 15, ..cycle };
+    let mut stale_revision = DataUsageCacheRevision::Etag("memory-2".to_string());
+    let stale_ctx = CancellationToken::new();
+    assert!(!persist_scanner_cycle_state(&stale_ctx, store, &mut stale_cycle, &mut stale_revision, 1,).await);
 }
 
 #[tokio::test]

crates/scanner/src/scanner_budget.rs

@@ -14,17 +14,16 @@
 
 use std::sync::{
     Arc,
-    atomic::{AtomicU8, AtomicU64, Ordering},
+    atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering},
 };
-use std::time::Instant;
-
-use tokio::time::Duration;
+use tokio::time::{Duration, Instant};
 use tokio_util::sync::CancellationToken;
 
 const BUDGET_REASON_NONE: u8 = 0;
 const BUDGET_REASON_RUNTIME: u8 = 1;
 const BUDGET_REASON_OBJECTS: u8 = 2;
 const BUDGET_REASON_DIRECTORIES: u8 = 3;
+const PROGRESS_CLOCK_SAMPLE_INTERVAL: u64 = 128;
 
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub(crate) struct ScannerCycleBudgetConfig {
@@ -63,29 +62,51 @@ pub struct ScannerCycleBudget {
     token: CancellationToken,
     reason: Arc<AtomicU8>,
     started_at: Instant,
+    deadline: Option<Instant>,
     max_duration: Option<Duration>,
     max_objects: Option<u64>,
     max_directories: Option<u64>,
     track_progress: bool,
+    track_unbounded_counts: bool,
     objects_scanned: AtomicU64,
     directories_started: AtomicU64,
     entries_visited: AtomicU64,
+    last_progress_millis: AtomicU64,
+    cycle_state_persisted: AtomicBool,
 }
 
 impl ScannerCycleBudget {
+    #[cfg(test)]
     pub(crate) fn new(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc<Self> {
-        Self::new_inner(parent, config, false)
+        Self::new_inner(parent, config, false, false)
     }
 
     pub(crate) fn new_with_progress_tracking(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc<Self> {
-        Self::new_inner(parent, config, true)
+        Self::new_inner(parent, config, true, true)
     }
 
-    fn new_inner(parent: &CancellationToken, config: ScannerCycleBudgetConfig, track_progress: bool) -> Arc<Self> {
+    pub(crate) fn new_with_runtime_progress_tracking(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc<Self> {
+        let track_progress = config.max_duration.is_some();
+        Self::new_inner(parent, config, track_progress, false)
+    }
+
+    fn new_inner(
+        parent: &CancellationToken,
+        config: ScannerCycleBudgetConfig,
+        track_progress: bool,
+        track_unbounded_counts: bool,
+    ) -> Arc<Self> {
         let token = parent.child_token();
         let reason = Arc::new(AtomicU8::new(BUDGET_REASON_NONE));
+        let started_at = Instant::now();
+        let deadline = config.max_duration.map(|duration| match started_at.checked_add(duration) {
+            Some(deadline) => deadline,
+            // Runtime config rejects this range, but keep programmatic callers
+            // fail-closed instead of panicking or silently disabling the wall clock.
+            None => started_at,
+        });
 
-        if let Some(duration) = config.max_duration {
+        if let Some(deadline) = deadline {
             let parent = parent.clone();
             let token_wait = token.clone();
             let token_cancel = token.clone();
@@ -94,7 +115,7 @@ impl ScannerCycleBudget {
                 tokio::select! {
                     _ = parent.cancelled() => {}
                     _ = token_wait.cancelled() => {}
-                    _ = tokio::time::sleep(duration) => {
+                    _ = tokio::time::sleep_until(deadline) => {
                         Self::cancel_for_reason(&reason, &token_cancel, ScannerCycleBudgetReason::Runtime);
                     }
                 }
@@ -104,14 +125,18 @@ impl ScannerCycleBudget {
         Arc::new(Self {
             token,
             reason,
-            started_at: Instant::now(),
+            started_at,
+            deadline,
             max_duration: config.max_duration,
             max_objects: config.max_objects,
             max_directories: config.max_directories,
             track_progress,
+            track_unbounded_counts,
             objects_scanned: AtomicU64::new(0),
             directories_started: AtomicU64::new(0),
             entries_visited: AtomicU64::new(0),
+            last_progress_millis: AtomicU64::new(0),
+            cycle_state_persisted: AtomicBool::new(false),
         })
     }
 
@@ -131,6 +156,14 @@ impl ScannerCycleBudget {
         self.max_duration
     }
 
+    pub(crate) fn deadline(&self) -> Option<Instant> {
+        self.deadline
+    }
+
+    pub(crate) fn cancel_for_runtime(&self) {
+        self.cancel_for(ScannerCycleBudgetReason::Runtime);
+    }
+
     pub(crate) fn max_objects(&self) -> Option<u64> {
         self.max_objects
     }
@@ -173,15 +206,43 @@ impl ScannerCycleBudget {
         self.entries_visited.load(Ordering::Relaxed)
     }
 
+    pub(crate) fn mark_cycle_state_persisted(&self) {
+        self.cycle_state_persisted.store(true, Ordering::Release);
+    }
+
+    pub(crate) fn cycle_state_persisted(&self) -> bool {
+        self.cycle_state_persisted.load(Ordering::Acquire)
+    }
+
+    pub(crate) fn progress_age(&self) -> Duration {
+        let elapsed_millis = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
+        let last_progress = self.last_progress_millis.load(Ordering::Relaxed);
+        Duration::from_millis(elapsed_millis.saturating_sub(last_progress))
+    }
+
+    fn record_progress_sample(&self, event: u64) {
+        // Clock reads are sampled at batch/count boundaries; the scanner's
+        // per-object path does not add a second progress atomic.
+        if event == 0 || (event != 1 && !event.is_multiple_of(PROGRESS_CLOCK_SAMPLE_INTERVAL)) {
+            return;
+        }
+        let elapsed_millis = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
+        self.last_progress_millis.store(elapsed_millis, Ordering::Relaxed);
+    }
+
     pub(crate) fn record_entries_visited(&self, entries_visited: u64) {
         if self.track_progress {
-            saturating_fetch_add(&self.entries_visited, entries_visited);
+            let entries = saturating_fetch_add(&self.entries_visited, entries_visited);
+            self.record_progress_sample(entries);
         }
     }
 
     pub(crate) fn record_remote_progress(&self, objects_scanned: u64, directories_started: u64) {
         if self.track_progress || self.max_objects.is_some() {
             let objects = saturating_fetch_add(&self.objects_scanned, objects_scanned);
+            if self.track_progress {
+                self.record_progress_sample(objects);
+            }
             if self.max_objects.is_some_and(|max_objects| objects >= max_objects) {
                 self.cancel_for(ScannerCycleBudgetReason::Objects);
             }
@@ -189,9 +250,12 @@ impl ScannerCycleBudget {
 
         if self.track_progress || self.max_directories.is_some() {
             let directories = saturating_fetch_add(&self.directories_started, directories_started);
+            if self.track_progress {
+                self.record_progress_sample(directories);
+            }
             if self
                 .max_directories
-                .is_some_and(|max_directories| directories > max_directories)
+                .is_some_and(|max_directories| directory_budget_exhausted(directories, max_directories))
             {
                 self.cancel_for(ScannerCycleBudgetReason::Directories);
             }
@@ -207,14 +271,17 @@ impl ScannerCycleBudget {
     }
 
     pub(crate) fn try_start_directory(&self) -> bool {
-        if !self.track_progress && self.max_directories.is_none() {
+        if self.max_directories.is_none() && !self.track_unbounded_counts {
             return true;
         }
 
         let directories = saturating_fetch_add(&self.directories_started, 1);
+        if self.track_progress {
+            self.record_progress_sample(directories);
+        }
         if self
             .max_directories
-            .is_some_and(|max_directories| directories > max_directories)
+            .is_some_and(|max_directories| directory_budget_exhausted(directories, max_directories))
         {
             self.cancel_for(ScannerCycleBudgetReason::Directories);
             return false;
@@ -224,11 +291,14 @@ impl ScannerCycleBudget {
     }
 
     pub(crate) fn record_object_scanned(&self) {
-        if !self.track_progress && self.max_objects.is_none() {
+        if self.max_objects.is_none() && !self.track_unbounded_counts {
             return;
         }
 
         let objects = saturating_fetch_add(&self.objects_scanned, 1);
+        if self.track_progress {
+            self.record_progress_sample(objects);
+        }
         if self.max_objects.is_some_and(|max_objects| objects >= max_objects) {
             self.cancel_for(ScannerCycleBudgetReason::Objects);
         }
@@ -259,6 +329,13 @@ fn saturating_fetch_add(value: &AtomicU64, delta: u64) -> u64 {
     }
 }
 
+fn directory_budget_exhausted(directories: u64, max_directories: u64) -> bool {
+    // Saturation hides a remote max+1 update when the configured limit is the
+    // largest representable counter. Treat that boundary as exhausted rather
+    // than allowing work to continue indefinitely.
+    directories > max_directories || (directories == u64::MAX && max_directories == u64::MAX)
+}
+
 impl Drop for ScannerCycleBudget {
     fn drop(&mut self) {
         self.token.cancel();
@@ -401,6 +478,35 @@ mod tests {
         assert_eq!(directory_budget.reason(), Some(ScannerCycleBudgetReason::Directories));
     }
 
+    #[test]
+    fn directory_budget_fails_closed_when_progress_saturates() {
+        let parent = CancellationToken::new();
+        let budget = ScannerCycleBudget::new(
+            &parent,
+            ScannerCycleBudgetConfig {
+                max_directories: Some(u64::MAX),
+                ..Default::default()
+            },
+        );
+
+        budget.record_remote_progress(0, u64::MAX);
+
+        assert_eq!(budget.reason(), Some(ScannerCycleBudgetReason::Directories));
+        assert!(budget.token().is_cancelled());
+
+        let local_budget = ScannerCycleBudget::new(
+            &parent,
+            ScannerCycleBudgetConfig {
+                max_directories: Some(u64::MAX),
+                ..Default::default()
+            },
+        );
+        local_budget.record_remote_progress(0, u64::MAX - 1);
+        assert!(!local_budget.budget_elapsed());
+        assert!(!local_budget.try_start_directory());
+        assert_eq!(local_budget.reason(), Some(ScannerCycleBudgetReason::Directories));
+    }
+
     #[test]
     fn explicit_progress_tracking_counts_unbounded_remote_work_without_cancelling() {
         let parent = CancellationToken::new();
@@ -461,4 +567,29 @@ mod tests {
         assert!(object_limited.requires_serial_progress_accounting());
         assert!(directory_limited.requires_serial_progress_accounting());
     }
+
+    #[tokio::test(start_paused = true)]
+    async fn progress_age_uses_virtual_time_and_sampled_progress() {
+        let parent = CancellationToken::new();
+        let budget = ScannerCycleBudget::new_with_runtime_progress_tracking(
+            &parent,
+            ScannerCycleBudgetConfig {
+                max_duration: Some(Duration::from_secs(60)),
+                ..Default::default()
+            },
+        );
+
+        tokio::time::advance(Duration::from_secs(5)).await;
+        assert_eq!(budget.progress_age(), Duration::from_secs(5));
+        budget.record_entries_visited(1);
+        assert_eq!(budget.progress_age(), Duration::ZERO);
+
+        tokio::time::advance(Duration::from_secs(2)).await;
+        for _ in 0..126 {
+            budget.record_entries_visited(1);
+        }
+        assert_eq!(budget.progress_age(), Duration::from_secs(2));
+        budget.record_entries_visited(1);
+        assert_eq!(budget.progress_age(), Duration::ZERO);
+    }
 }

crates/scanner/src/scanner_folder.rs

@@ -55,9 +55,9 @@ use tracing::{debug, error, warn};
 use crate::{
     Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts, ReplicationConfig,
     ReplicationHealObject, ReplicationQueueAdmission, ReplicationStatusType, STORAGE_FORMAT_FILE, ScannerDiskExt as _,
-    ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule,
-    enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object,
-    path2_bucket_object_with_base_path, queue_replication_heal, scanner_is_erasure,
+    ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, TierRegistrySnapshot, apply_expiry_rule,
+    apply_transition_rule, enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object,
+    path2_bucket_object_with_base_path, queue_replication_heal, runtime_tier_registry, scanner_is_erasure,
     scanner_replication_config_for_lifecycle_eval,
 };
 use crate::{ScannerObjectInfo as ObjectInfo, ScannerObjectToDelete as ObjectToDelete};
@@ -626,6 +626,7 @@ fn apply_scanner_size_summary(into: &mut DataUsageEntry, summary: &SizeSummary)
     }
 
     into.add_tier_sizes(&summary.tier_stats);
+    into.add_unknown_tier_stats(&summary.unknown_tier_stats);
 }
 
 fn data_usage_root_has_progress(root: &DataUsageEntry) -> bool {
@@ -670,6 +671,9 @@ pub struct FolderScanner {
     budget: Arc<ScannerCycleBudget>,
     skip_heal: Arc<std::sync::atomic::AtomicBool>,
     local_disk: Arc<Disk>,
+    /// Tier registry frozen for this folder scan. A refresh applies to the
+    /// next scan and cannot mix generations in one aggregate.
+    tier_registry: TierRegistrySnapshot,
     pending_heals_changed: bool,
     #[cfg(test)]
     list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
@@ -1329,7 +1333,11 @@ impl FolderScanner {
                     continue;
                 }
 
-                let sz = match self.local_disk.get_size(item.clone()).await {
+                let sz = match self
+                    .local_disk
+                    .get_size_with_tier_names(item.clone(), &self.tier_registry.names)
+                    .await
+                {
                     Ok(sz) => sz,
                     Err(e) => {
                         let failure_action = classify_get_size_failure(&item, &e);
@@ -2161,6 +2169,10 @@ pub async fn scan_data_folder(
 
     let failed_object_ttl = rustfs_utils::get_env_u32(ENV_FAILED_OBJECT_TTL_SECS, DEFAULT_FAILED_OBJECT_TTL_SECS) as u64;
     let failed_objects_max = rustfs_utils::get_env_u32(ENV_FAILED_OBJECTS_MAX, DEFAULT_FAILED_OBJECTS_MAX) as usize;
+    let tier_registry = runtime_tier_registry().await;
+    let mut cache = cache;
+    cache.fold_retired_tiers(&tier_registry.names);
+    cache.info.tier_registry_generation = Some(tier_registry.generation);
 
     // Create folder scanner
     let mut scanner = FolderScanner {
@@ -2189,6 +2201,7 @@ pub async fn scan_data_folder(
         budget: budget.clone(),
         skip_heal,
         local_disk,
+        tier_registry,
         pending_heals_changed: false,
         #[cfg(test)]
         list_path_raw_options_observer: None,

crates/scanner/src/scanner_folder/item_actions.rs

@@ -289,11 +289,42 @@ impl ScannerItem {
         item.object_path()
     }
 
+    fn effective_tier(oi: &ObjectInfo) -> &str {
+        if oi.transitioned_object.status == crate::TRANSITION_COMPLETE {
+            oi.transitioned_object.tier.as_str()
+        } else {
+            oi.storage_class.as_deref().unwrap_or(crate::storageclass::STANDARD)
+        }
+    }
+
+    fn tier_is_known(oi: &ObjectInfo, tier_names: &[String]) -> bool {
+        let tier = Self::effective_tier(oi);
+        if tier == crate::data_usage_define::UNKNOWN_TIER {
+            return false;
+        }
+        tier == crate::storageclass::STANDARD || tier == crate::storageclass::RRS || tier_names.iter().any(|name| name == tier)
+    }
+
+    fn action_requires_known_tier(action: IlmAction) -> bool {
+        matches!(
+            action,
+            IlmAction::TransitionAction
+                | IlmAction::TransitionVersionAction
+                | IlmAction::DeleteAction
+                | IlmAction::DeleteVersionAction
+                | IlmAction::DeleteRestoredAction
+                | IlmAction::DeleteRestoredVersionAction
+                | IlmAction::DeleteAllVersionsAction
+                | IlmAction::DelMarkerDeleteAllVersionsAction
+        )
+    }
+
     pub async fn apply_actions(
         &mut self,
         object_infos: Vec<ObjectInfo>,
         lock_retention: Option<Arc<ObjectLockConfiguration>>,
         versioning_config: VersioningConfiguration,
+        tier_names: &[String],
         size_summary: &mut SizeSummary,
     ) {
         let object_path = self.object_path();
@@ -425,6 +456,18 @@ impl ScannerItem {
                 let mut size = actual_size;
                 let mut account_now = true;
 
+                // A retired/unknown source tier may point at a remote object
+                // that cannot be safely deleted or transitioned. Lifecycle
+                // evaluation is still useful for accounting, but all
+                // side-effecting tier actions fail closed until the registry
+                // recognizes the source again.
+                if !Self::tier_is_known(oi, tier_names) && Self::action_requires_known_tier(event.action) {
+                    size = self.heal_actions(oi, actual_size, size_summary).await;
+                    size_summary.actions_accounting(oi, size, actual_size);
+                    cumulative_size += size;
+                    continue;
+                }
+
                 match event.action {
                     IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => {
                         debug!(
@@ -929,4 +972,15 @@ mod tests {
         assert_eq!(item.object_name, "object");
         assert_eq!(item.object_path(), "object");
     }
+
+    #[test]
+    fn unknown_tier_never_triggers_transition() {
+        let object = ObjectInfo {
+            storage_class: Some("retired-tier".to_string()),
+            ..Default::default()
+        };
+        assert!(!ScannerItem::tier_is_known(&object, &["WARM".to_string()]));
+        assert!(ScannerItem::action_requires_known_tier(IlmAction::TransitionAction));
+        assert!(ScannerItem::action_requires_known_tier(IlmAction::DeleteVersionAction));
+    }
 }

crates/scanner/src/scanner_folder/tests.rs

@@ -325,6 +325,11 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         budget: ScannerCycleBudget::new(&CancellationToken::new(), Default::default()),
         skip_heal: Arc::new(AtomicBool::new(false)),
         local_disk: disk,
+        tier_registry: crate::TierRegistrySnapshot {
+            generation: 0,
+            names: Arc::new([]),
+            refresh_failed: false,
+        },
         pending_heals_changed: false,
         list_path_raw_options_observer: None,
     };

crates/scanner/src/scanner_io.rs

@@ -48,6 +48,7 @@ use time::OffsetDateTime;
 use tokio::sync::{Mutex, Notify, Semaphore, mpsc};
 use tokio::time::Duration;
 use tokio_util::sync::CancellationToken;
+use tokio_util::task::AbortOnDropHandle;
 use tracing::{debug, error, warn};
 
 use crate::ScannerObjectInfo as ObjectInfo;
@@ -497,6 +498,9 @@ pub trait ScannerIODisk: Send + Sync + Debug + 'static {
     ) -> Result<ScannerDiskScanOutcome>;
 
     async fn get_size(&self, item: ScannerItem) -> Result<SizeSummary>;
+
+    /// Read one object using a registry snapshot captured at scan start.
+    async fn get_size_with_tier_names(&self, item: ScannerItem, tier_names: &[String]) -> Result<SizeSummary>;
 }
 
 #[derive(Debug)]

crates/scanner/src/scanner_io/cache.rs

@@ -216,6 +216,7 @@ pub(super) fn completed_data_usage_info(
         delete_markers_total_count: u64::try_from(total.delete_markers).ok()?,
         objects_total_size: u64::try_from(total.size).ok()?,
         tier_stats: total.all_tier_stats.filter(|tiers| !tiers.is_empty()),
+        unknown_tier_stats: total.unknown_tier_stats.filter(|stats| !stats.is_empty()),
         buckets_count: u64::try_from(all_buckets.len()).ok()?,
         bucket_sizes,
         buckets_usage,

crates/scanner/src/scanner_io/io_cache.rs

@@ -314,7 +314,7 @@ impl ScannerIOCache for SetDisks {
         let ctx_clone = ctx.clone();
         let completed_bucket_count = Arc::new(AtomicUsize::new(0));
         let completed_bucket_count_clone = completed_bucket_count.clone();
-        let collect_bucket_results_fut = tokio::spawn(async move {
+        let collect_bucket_results_fut = AbortOnDropHandle::new(tokio::spawn(async move {
             let mut cancelled = false;
 
             loop {
@@ -333,7 +333,7 @@ impl ScannerIOCache for SetDisks {
                     }
                 }
             }
-        });
+        }));
 
         let mut futs = Vec::new();
 
@@ -365,7 +365,7 @@ impl ScannerIOCache for SetDisks {
                 NamespaceScannerWorkerMode::RemoteV4(server_epoch) => Some(server_epoch),
                 NamespaceScannerWorkerMode::Coordinator => None,
             };
-            futs.push(tokio::spawn(async move {
+            futs.push(AbortOnDropHandle::new(tokio::spawn(async move {
                 let remote_session_id = uuid::Uuid::new_v4();
                 let mut remote_session_sequence = 0_u64;
                 loop {
@@ -1038,7 +1038,7 @@ impl ScannerIOCache for SetDisks {
                         );
                     }
                 }
-            }));
+            })));
         }
         drop(bucket_tx);
         drop(bucket_result_tx);

crates/scanner/src/scanner_io/io_cycle.rs

@@ -242,7 +242,7 @@ impl ScannerIOCycle for ECStore {
                         results[results_index_clone] = result;
                     }
                 });
-                wait_futs.push(receiver_fut);
+                wait_futs.push(AbortOnDropHandle::new(receiver_fut));
 
                 let scan_plan = ScannerBucketScanPlan {
                     buckets: set_buckets,
@@ -318,7 +318,7 @@ impl ScannerIOCycle for ECStore {
                         record_set_scan_failure(&mut first_err, e);
                     }
                 });
-                wait_futs.push(scanner_fut);
+                wait_futs.push(AbortOnDropHandle::new(scanner_fut));
             }
         }
 

crates/scanner/src/scanner_io/io_disk.rs

@@ -13,29 +13,38 @@
 // limitations under the License.
 /// ScannerIODisk implementation for Disk: get_size and the per-disk bucket scan.
 use super::*;
+use crate::UNKNOWN_TIER;
 
 ///
 /// Seed [`SizeSummary::tier_stats`] from the cached tier-name list.
 ///
-/// Preserves the original seeding semantics: with no tiers configured the map
-/// stays completely empty (STANDARD/RRS are not seeded either); otherwise the
-/// standard storage classes are seeded alongside every configured tier so
-/// per-object accounting always finds its tier key.
+/// Preserves the original no-tier shape: with no tiers configured the map
+/// stays completely empty (STANDARD/RRS/UNKNOWN are not seeded either).
+/// Otherwise the standard storage classes and one fixed unknown bucket are
+/// seeded alongside every configured tier so per-object accounting never
+/// inserts an untrusted metadata key.
 pub(super) fn tier_stats_template(tier_names: &[String]) -> HashMap<String, TierStats> {
-    let mut tier_stats = HashMap::with_capacity(tier_names.len() + 2);
+    let mut tier_stats = HashMap::with_capacity(tier_names.len() + 3);
     for tier_name in tier_names {
-        tier_stats.insert(tier_name.clone(), TierStats::default());
+        if tier_name != UNKNOWN_TIER {
+            tier_stats.insert(tier_name.clone(), TierStats::default());
+        }
     }
     if !tier_stats.is_empty() {
         tier_stats.insert(storageclass::STANDARD.to_string(), TierStats::default());
         tier_stats.insert(storageclass::RRS.to_string(), TierStats::default());
+        tier_stats.insert(UNKNOWN_TIER.to_string(), TierStats::default());
     }
     tier_stats
 }
 
 #[async_trait::async_trait]
 impl ScannerIODisk for Disk {
-    async fn get_size(&self, mut item: ScannerItem) -> Result<SizeSummary> {
+    async fn get_size(&self, item: ScannerItem) -> Result<SizeSummary> {
+        self.get_size_with_tier_names(item, &runtime_tier_names().await).await
+    }
+
+    async fn get_size_with_tier_names(&self, mut item: ScannerItem, tier_names: &[String]) -> Result<SizeSummary> {
         let done_object = Metrics::time(Metric::ScanObject);
 
         if !is_xl_meta_path(&item.path) {
@@ -107,10 +116,10 @@ impl ScannerIODisk for Disk {
 
         let mut size_summary = SizeSummary::default();
 
-        // Tier names come from the process-wide TTL cache; seeding from them
-        // replaces the per-object clone of every full TierConfig.
-        let tier_names = runtime_tier_names().await;
-        size_summary.tier_stats = tier_stats_template(&tier_names);
+        // The caller supplies one registry snapshot for the whole folder scan;
+        // seeding from it prevents a TTL refresh from mixing generations in a
+        // single result.
+        size_summary.tier_stats = tier_stats_template(tier_names);
 
         let lock_config = object_lock_config_for_scanner_item(&item).await;
 
@@ -120,7 +129,7 @@ impl ScannerIODisk for Disk {
         // `object_infos`.
         global_metrics().record_scanner_versions_scanned(object_infos.len() as u64);
 
-        item.apply_actions(object_infos, lock_config, versioning_config, &mut size_summary)
+        item.apply_actions(object_infos, lock_config, versioning_config, tier_names, &mut size_summary)
             .await;
 
         if !free_version_infos.is_empty() {

crates/scanner/src/scanner_io/tests.rs

@@ -23,7 +23,7 @@ use crate::storage_api::owner::{
 use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _};
 use crate::{
     DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerObjectOptions,
-    ScannerPutObjReader, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests,
+    ScannerPutObjReader, UNKNOWN_TIER, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests,
     init_local_disks_with_instance_ctx, new_disk, path2_bucket_object_with_base_path,
 };
 use rustfs_filemeta::FileInfo;
@@ -986,8 +986,8 @@ fn is_xl_meta_path_accepts_forward_separator() {
 fn tier_stats_template_seeds_tiers_and_standard_classes() {
     let template = tier_stats_template(&["WARM".to_string(), "COLD".to_string()]);
 
-    assert_eq!(template.len(), 4);
-    for tier in ["WARM", "COLD", storageclass::STANDARD, storageclass::RRS] {
+    assert_eq!(template.len(), 5);
+    for tier in ["WARM", "COLD", storageclass::STANDARD, storageclass::RRS, UNKNOWN_TIER] {
         assert_eq!(template.get(tier), Some(&TierStats::default()), "missing seed for tier {tier}");
     }
 }

crates/utils/src/envs.rs

@@ -268,7 +268,7 @@ where
         .parse::<T>()
         .map_err(|_| {
             log_once(&format!("env_invalid_value:{used_key}"), || {
-                format!("Invalid {} value for {used_key}: {value}. Treating as unset.", type_name::<T>())
+                format!("Invalid {} value for {used_key}. Treating as unset.", type_name::<T>())
             });
         })
         .ok()
@@ -570,7 +570,7 @@ where
         Ok(parsed) => EnvParseOutcome::Parsed(parsed),
         Err(_) => {
             log_once(&format!("env_invalid_value:{used_key}"), || {
-                format!("Invalid {} value for {used_key}: {value}. Treating as unset.", type_name::<T>())
+                format!("Invalid {} value for {used_key}. Treating as unset.", type_name::<T>())
             });
             EnvParseOutcome::Invalid
         }

docs/operations/scanner-runtime-controls.md

@@ -52,7 +52,7 @@ The `/v3/scanner/status` response reports each effective runtime value with a
 | `scanner.max_wait` | `RUSTFS_SCANNER_MAX_WAIT_SECS` | seconds | preset-derived | Caps one scanner sleep. |
 | `scanner.cycle` | `RUSTFS_SCANNER_CYCLE` | seconds | preset-derived | Sets the interval between scanner cycles. |
 | `scanner.start_delay` | `RUSTFS_SCANNER_START_DELAY_SECS` | seconds | unset | Sets startup delay and, for compatibility, the cycle interval when `scanner.cycle` is unset. |
-| `scanner.cycle_max_duration` | `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` | seconds | `0` | Caps one cycle's runtime. `0` disables this budget. |
+| `scanner.cycle_max_duration` | `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` | seconds | `1800` | Caps one cycle's runtime. An explicit `0` disables this budget. |
 | `scanner.cycle_max_objects` | `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` | objects | `0` | Caps objects processed by one cycle. `0` disables this budget. |
 | `scanner.cycle_max_directories` | `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` | directories | `0` | Caps directories entered by one cycle. `0` disables this budget. |
 | `heal.bitrot_cycle` | `RUSTFS_SCANNER_BITROT_CYCLE_SECS` | seconds | `2592000` | Controls periodic deep bitrot scans. `false`, `off`, `no`, or `disabled` disables periodic deep scans; `0`, `true`, `on`, or `yes` runs deep mode every scanner cycle. |
@@ -70,6 +70,21 @@ sleep multiplier, maximum wait, and cycle interval. Use `scanner.delay`,
 `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis
 needs a precise override.
 
+When the cycle duration control is unset, RustFS uses a finite 1800-second
+(30-minute) default, matching the scanner benchmark guidance. An explicit `0`
+preserves the compatibility behavior of an unbounded cycle; object and
+directory budgets likewise remain unbounded when explicitly set to `0`. Invalid
+or overflowing duration environment values are configuration errors rather than
+silent fallback values.
+
+When a finite deadline expires, RustFS cancels cooperative scanner work and
+waits only for the existing bounded shutdown window. A non-yielding I/O future
+is dropped after that window. RustFS then attempts a higher leadership epoch so
+late cycle, usage, cache, and remote writes from the old generation fail closed.
+If the worker cannot stop cooperatively, the cycle state was not confirmed
+durable, or that epoch fence cannot be durably persisted, the scanner reports
+`recovery-required`; it does not claim an uncooperative cursor was saved.
+
 An explicit `scanner.cycle` or `RUSTFS_SCANNER_CYCLE` is a minimum inter-cycle
 cadence: dirty-usage notifications do not bypass that configured interval.
 The default adaptive policy continues to use dirty-usage notifications to wake
@@ -144,6 +159,10 @@ metrics.maintenance_control.primary_control
 metrics.source_work
 metrics.replication_repair
 metrics.scan_checkpoint
+metrics.cycle_timeout_total
+metrics.cycle_last_progress_age
+metrics.leader_lease_without_progress
+metrics.cycle_recovery_required_total
 ```
 
 ## Reading Pacing Pressure