feat(scanner): fair bucket service across restarts and budgeted first enumeration (#9083)

* feat(scanner): bound first-build enumeration with the cycle budget

Charge raw directory entries and their name bytes, and let one bucket scan hold a child budget that stops its own token while charging the cycle. A count budget also bounds one directory listing, so a wide first build can no longer finish a full read_dir before the first object is charged.

The new stop reason maps onto the persisted Directories variant so older readers keep decoding checkpoints. A directory with queued children keeps its index until they end, and every index on the active walk path is saved as an optional map field: losing either one would recharge its replay and stop a fixed budget from advancing.

* feat(scanner): keep bucket service order across restarts

Save each set's first-in-first-out bucket order as an optional map field of its usage cache, including after an incomplete cycle, which otherwise writes nothing. Durable age now outranks the dirty-first and leader-local cohort orders, which only break its ties, so a restart cannot hand the budget back to the same first buckets. Failed turns add visible debt without moving a bucket forward; a budget stop or a resource veto adds none. A bucket that spends its quantum and saves forward motion returns to the queue tail within the cycle.

The order authorizes nothing: another movement epoch or a corrupt value starts a new one, and a recreated bucket joins as a new arrival.

* test(scanner): cover quantum handoff and saved service order

Run the real worker queue with injected bucket shares and verify a spent bucket yields to waiters, a share without durable motion ends its turn, and the saved order keeps certified coverage while rejecting older owners.

* docs(scanner): describe budgeted first build and service order

* test(scanner): pin index retention and requeue motion

Bound the nested restart fixture tightly enough that dropping a directory index fails it, and cover a repeated share whose frontier never moves.

* fix(scanner): keep service generations and owner stops intact

Queue a new bucket behind waiters but ahead of buckets that already had a turn, matching the leader-local cohort, and keep the set cache untouched when the owner was cancelled so an interrupted cycle cannot advance its revision. The order holds one record per inventory bucket, like the set cache's entries, so a truncating bound that could starve the long-waiting buckets is removed. Publish the incomplete scope before saving so a contended lock cannot delay it.

* ci(e2e): refresh the full selection digests

* fix(scanner): correct a comment spelling

* fix(scanner): separate service debt from vetoes and stale owners

Keep a resource veto and a remote count-budget stop out of failure debt,
settle the claim before a requeue exposes the bucket to another worker, and
stamp the service order with its leader so an older owner cannot replace it
after an incomplete save. The checkpoint fixture's budget now covers both
service orders.

* fix(ecstore): create parents before resource-scoped writes

#9064 moved parent creation into the lazy writer, so a write that holds a
resource unit reached the native open without its parent directory and
failed with FileNotFound on nested paths.

* fix(e2e): retry SlowDown codes during upgrade write readiness

The rolling-upgrade readiness probe only retried 503 ServiceUnavailable,
so a transient 503 SlowDown from a still-recovering node failed the
four_node_rolling_upgrade test immediately. Accept SlowDown and
SlowDownRead alongside ServiceUnavailable, matching the existing chaos
test transient-error set.

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=67251cb7d9978d38e899268cac4d7898297ff7e5b5c48c9c1546dc760b93568d
-sha256-linux=673da76c2e5710c4622fe2fa32d2b297730895769487fcf1152ee1c2366f32f4
+sha256-darwin=5d4509d69c3106c1b8a6c353d123440d6b4dbe5dd42612e42dcb864946beae6c
+sha256-linux=414e1dae735e6d14821a992cebc09a545576e9a77e10a5e0ed671ab07b89df98

crates/e2e_test/src/upgrade_compatibility_test.rs

@@ -1112,8 +1112,9 @@ async fn wait_for_upgrade_writes(
             match response {
                 Ok(Ok(_)) => break,
                 Ok(Err(error)) => {
+                    let code = error.as_service_error().and_then(ProvideErrorMetadata::code);
                     if error.raw_response().map(|response| response.status().as_u16()) != Some(503)
-                        || error.as_service_error().and_then(ProvideErrorMetadata::code) != Some("ServiceUnavailable")
+                        || !matches!(code, Some("ServiceUnavailable" | "SlowDown" | "SlowDownRead"))
                     {
                         return Err(format!("{phase}: node {node} {probe} failed: {error:?}").into());
                     }

crates/ecstore/src/disk/local.rs

@@ -5331,6 +5331,9 @@ impl LocalIoBackend for StdBackend {
                 }
 
                 if let Some(resources) = current_resource_unit() {
+                    if let Some(parent) = file_path.parent() {
+                        os::make_dir_all(parent, &volume_dir).await?;
+                    }
                     return self.open_resource_writer(file_path, mode, resources).await;
                 }
 

crates/scanner/src/data_usage_define.rs

@@ -596,9 +596,205 @@ pub struct PendingScannerHeal {
     pub last_admission_reason: String,
 }
 
+/// One bucket's standing in its set's service order. The wait position, the
+/// quanta granted and the last certification are separate quantities: a bucket
+/// that keeps failing moves to the tail like any other and only adds debt.
+#[derive(Clone, Copy, Debug, Default, Deserialize, PartialEq, Eq)]
+pub struct DataUsageBucketService {
+    #[serde(default)]
+    pub waiting_since: u64,
+    #[serde(default)]
+    pub service_quanta: u64,
+    #[serde(default)]
+    pub last_certified: Option<u64>,
+    #[serde(default)]
+    pub failure_debt: u64,
+    #[serde(default)]
+    pub incarnation: Option<uuid::Uuid>,
+}
+
+// Nested derives encode positionally; a map lets later fields stay optional.
+impl Serialize for DataUsageBucketService {
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        let field_count = 2
+            + usize::from(self.last_certified.is_some())
+            + usize::from(self.failure_debt != 0)
+            + usize::from(self.incarnation.is_some());
+        let mut map = serializer.serialize_map(Some(field_count))?;
+        map.serialize_entry("waiting_since", &self.waiting_since)?;
+        map.serialize_entry("service_quanta", &self.service_quanta)?;
+        if let Some(last_certified) = self.last_certified {
+            map.serialize_entry("last_certified", &last_certified)?;
+        }
+        if self.failure_debt != 0 {
+            map.serialize_entry("failure_debt", &self.failure_debt)?;
+        }
+        if let Some(incarnation) = self.incarnation {
+            map.serialize_entry("incarnation", &incarnation)?;
+        }
+        map.end()
+    }
+}
+
+/// Durable first-in-first-out service order of one set, in generations: every
+/// bucket at or below `generation_floor` still waits for its turn and everything
+/// above it was served, so a new bucket queues behind waiters but ahead of
+/// buckets that already had a turn. It only orders work:
+/// no entry proves coverage, authorizes a skip or releases a responsibility,
+/// so losing it restarts fairness and nothing else. Like the set cache's bucket
+/// entries and incarnations, it holds one record per bucket of the inventory.
+#[derive(Clone, Debug, Default, Deserialize, PartialEq, Eq)]
+pub struct DataUsageBucketServiceLedger {
+    /// Movement epoch the order was built under; another epoch invalidates it.
+    #[serde(default)]
+    pub publication_epoch: u64,
+    /// Leader that saved the order; an incomplete save stamps nothing else.
+    #[serde(default)]
+    pub leader_epoch: u64,
+    #[serde(default)]
+    pub next_sequence: u64,
+    #[serde(default)]
+    pub generation_floor: u64,
+    #[serde(default)]
+    pub buckets: HashMap<String, DataUsageBucketService>,
+}
+
+impl Serialize for DataUsageBucketServiceLedger {
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        let mut map = serializer.serialize_map(Some(5))?;
+        map.serialize_entry("publication_epoch", &self.publication_epoch)?;
+        map.serialize_entry("leader_epoch", &self.leader_epoch)?;
+        map.serialize_entry("next_sequence", &self.next_sequence)?;
+        map.serialize_entry("generation_floor", &self.generation_floor)?;
+        map.serialize_entry("buckets", &self.buckets)?;
+        map.end()
+    }
+}
+
+impl DataUsageBucketServiceLedger {
+    /// Rebuilds the order for this cycle's inventory. Decoded state that fails
+    /// validation or belongs to another movement epoch starts a new order;
+    /// a bucket whose incarnation changed is a new arrival behind all waiters.
+    pub(crate) fn restore<'a>(
+        persisted: Option<Self>,
+        publication_epoch: u64,
+        inventory: impl Iterator<Item = &'a str>,
+        incarnations: Option<&HashMap<String, uuid::Uuid>>,
+    ) -> Self {
+        let mut ledger = persisted
+            .filter(|ledger| ledger.publication_epoch == publication_epoch && ledger.is_valid())
+            .unwrap_or_else(|| Self {
+                publication_epoch,
+                ..Default::default()
+            });
+        let previous = std::mem::take(&mut ledger.buckets);
+        let mut arrivals = Vec::new();
+        for bucket in inventory {
+            let incarnation = incarnations.and_then(|current| current.get(bucket)).copied();
+            match previous.get(bucket) {
+                Some(service) if service.incarnation.is_none() || incarnation.is_none() || service.incarnation == incarnation => {
+                    ledger.buckets.insert(
+                        bucket.to_owned(),
+                        DataUsageBucketService {
+                            incarnation: incarnation.or(service.incarnation),
+                            ..*service
+                        },
+                    );
+                }
+                _ => arrivals.push((bucket, incarnation)),
+            }
+        }
+        if arrivals.is_empty() {
+            let floor = ledger.generation_floor;
+            if ledger.buckets.values().all(|service| service.waiting_since > floor) {
+                ledger.generation_floor = ledger.next_sequence;
+            }
+            return ledger;
+        }
+        // Open one position between the waiters and the served buckets.
+        let floor = ledger.generation_floor;
+        ledger.generation_floor = floor.saturating_add(1);
+        ledger.next_sequence = ledger.next_sequence.saturating_add(1);
+        for service in ledger.buckets.values_mut().filter(|service| service.waiting_since > floor) {
+            service.waiting_since = service.waiting_since.saturating_add(1);
+        }
+        for (bucket, incarnation) in arrivals {
+            ledger.buckets.insert(
+                bucket.to_owned(),
+                DataUsageBucketService {
+                    waiting_since: ledger.generation_floor,
+                    incarnation,
+                    ..Default::default()
+                },
+            );
+        }
+        ledger
+    }
+
+    fn is_valid(&self) -> bool {
+        self.generation_floor <= self.next_sequence
+            && self
+                .buckets
+                .iter()
+                .all(|(bucket, service)| !bucket.is_empty() && service.waiting_since <= self.next_sequence)
+    }
+
+    /// Smaller positions have waited longer; an untracked bucket queues last.
+    pub(crate) fn wait_position(&self, bucket: &str) -> u64 {
+        self.buckets.get(bucket).map_or(u64::MAX, |service| service.waiting_since)
+    }
+
+    /// Moves a bucket to the tail and returns its previous wait position.
+    pub(crate) fn record_service(&mut self, bucket: &str) -> Option<u64> {
+        let service = self.buckets.get_mut(bucket)?;
+        self.next_sequence = self.next_sequence.saturating_add(1);
+        let previous = std::mem::replace(&mut service.waiting_since, self.next_sequence);
+        service.service_quanta = service.service_quanta.saturating_add(1);
+        Some(previous)
+    }
+
+    /// Undoes `record_service` for an admission a resource veto turned back.
+    pub(crate) fn return_service(&mut self, bucket: &str, waiting_since: u64) {
+        if let Some(service) = self.buckets.get_mut(bucket) {
+            service.waiting_since = waiting_since;
+            service.service_quanta = service.service_quanta.saturating_sub(1);
+        }
+    }
+
+    pub(crate) fn record_certified(&mut self, bucket: &str) {
+        if let Some(service) = self.buckets.get_mut(bucket) {
+            service.last_certified = Some(service.waiting_since);
+            service.failure_debt = 0;
+        }
+    }
+
+    pub(crate) fn record_failure(&mut self, bucket: &str) {
+        if let Some(service) = self.buckets.get_mut(bucket) {
+            service.failure_debt = service.failure_debt.saturating_add(1);
+        }
+    }
+
+    /// Buckets holding unpaid failures and the first of them by name.
+    pub(crate) fn failure_debt(&self) -> (usize, Option<&str>) {
+        let mut debtors = self
+            .buckets
+            .iter()
+            .filter(|(_, service)| service.failure_debt > 0)
+            .map(|(bucket, _)| bucket.as_str());
+        let first = debtors.next();
+        let (count, first) = debtors.fold((usize::from(first.is_some()), first), |(count, first), bucket| {
+            (count + 1, first.min(Some(bucket)))
+        });
+        (count, first)
+    }
+}
+
 /// Data usage cache info
 #[derive(Clone, Debug, Default, Deserialize)]
 pub struct DataUsageCacheInfo {
+    /// In-memory only: this partial stopped on admission pressure, not a failure.
+    #[serde(skip)]
+    pub scan_resource_deferred: bool,
     pub name: String,
     pub next_cycle: u64,
     pub last_update: Option<SystemTime>,
@@ -615,6 +811,11 @@ pub struct DataUsageCacheInfo {
     pub scan_raw_enumeration_cursor: Option<DataUsageRawEnumerationCursor>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub scan_raw_enumeration_page_index: Option<RawEnumerationPageIndex>,
+    /// Indexes of the other directories on the active walk path. A restart
+    /// must replay every wide ancestor before it reaches the selected one, and
+    /// recharging those replays would stop a fixed budget from ever advancing.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub scan_raw_enumeration_path_page_indexes: Vec<RawEnumerationPageIndex>,
     #[serde(default)]
     pub scan_identity: Option<DataUsageScanIdentity>,
     #[serde(default)]
@@ -674,6 +875,9 @@ pub struct DataUsageCacheInfo {
     /// skipping an unselected bucket in a later scoped set scan.
     #[serde(default, skip_serializing_if = "HashMap::is_empty")]
     pub scan_bucket_incarnations: HashMap<String, uuid::Uuid>,
+    /// Bucket service order of a set cache; absent from bucket caches.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub scan_bucket_service: Option<DataUsageBucketServiceLedger>,
 }
 
 impl Serialize for DataUsageCacheInfo {
@@ -686,6 +890,7 @@ impl Serialize for DataUsageCacheInfo {
         let field_count = 16
             + usize::from(self.scan_raw_enumeration_cursor.is_some())
             + usize::from(self.scan_raw_enumeration_page_index.is_some())
+            + usize::from(!self.scan_raw_enumeration_path_page_indexes.is_empty())
             + usize::from(self.scan_identity.is_some())
             + usize::from(self.scan_progress.is_some())
             + usize::from(self.scan_coverage_receipt.is_some())
@@ -699,7 +904,8 @@ impl Serialize for DataUsageCacheInfo {
             + usize::from(self.lkg_scan_plan_digest.is_some())
             + usize::from(self.scan_execution_digest.is_some())
             + usize::from(self.segment_invalidation_proof.is_some())
-            + usize::from(!self.scan_bucket_incarnations.is_empty());
+            + usize::from(!self.scan_bucket_incarnations.is_empty())
+            + usize::from(self.scan_bucket_service.is_some());
         let mut state = serializer.serialize_map(Some(field_count))?;
         state.serialize_entry("name", &self.name)?;
         state.serialize_entry("next_cycle", &self.next_cycle)?;
@@ -717,6 +923,9 @@ impl Serialize for DataUsageCacheInfo {
         if let Some(index) = &self.scan_raw_enumeration_page_index {
             state.serialize_entry("scan_raw_enumeration_page_index", index)?;
         }
+        if !self.scan_raw_enumeration_path_page_indexes.is_empty() {
+            state.serialize_entry("scan_raw_enumeration_path_page_indexes", &self.scan_raw_enumeration_path_page_indexes)?;
+        }
         if let Some(identity) = self.scan_identity {
             state.serialize_entry("scan_identity", &identity)?;
         }
@@ -765,6 +974,9 @@ impl Serialize for DataUsageCacheInfo {
         if !self.scan_bucket_incarnations.is_empty() {
             state.serialize_entry("scan_bucket_incarnations", &self.scan_bucket_incarnations)?;
         }
+        if let Some(ledger) = &self.scan_bucket_service {
+            state.serialize_entry("scan_bucket_service", ledger)?;
+        }
         state.end()
     }
 }
@@ -902,6 +1114,30 @@ impl DataUsageCache {
         }
     }
 
+    /// Identifies how far durable resume state has advanced. Discovery alone
+    /// does not change it; only committed raw pages or coverage frontier do.
+    pub(crate) fn durable_scan_position(&self) -> (Option<String>, Option<String>, u64) {
+        let raw_entries = self
+            .info
+            .scan_raw_enumeration_page_index
+            .iter()
+            .chain(&self.info.scan_raw_enumeration_path_page_indexes)
+            .map(|index| match index.status() {
+                RawEnumerationPageOwnerStatus::Unsupported => 0,
+                RawEnumerationPageOwnerStatus::Building { indexed_entries, .. }
+                | RawEnumerationPageOwnerStatus::Ready { indexed_entries, .. } => indexed_entries,
+            })
+            .fold(0, u64::saturating_add);
+        (
+            self.info.scan_resume_after.clone(),
+            self.info
+                .scan_coverage_receipt
+                .as_ref()
+                .map(|receipt| receipt.through.clone()),
+            raw_entries,
+        )
+    }
+
     pub(crate) fn has_scan_position(&self) -> bool {
         self.info.scan_resume_after.is_some()
             || self.info.scan_checkpoint.is_some()
@@ -1044,7 +1280,14 @@ impl DataUsageCache {
         }
         if self.validated_raw_enumeration_page_index().is_none() {
             self.info.scan_raw_enumeration_page_index = None;
+            self.info.scan_raw_enumeration_path_page_indexes.clear();
         }
+        let path_indexes = std::mem::take(&mut self.info.scan_raw_enumeration_path_page_indexes);
+        self.info.scan_raw_enumeration_path_page_indexes = path_indexes
+            .into_iter()
+            .take(MAX_DATA_USAGE_CACHE_DEPTH)
+            .filter(|index| self.raw_enumeration_page_index_is_valid(index))
+            .collect();
         let cursor_is_valid = !self.has_scan_position()
             || self.validated_scan_frontier().is_some()
             || self.info.scan_raw_enumeration_cursor.is_some()
@@ -1087,6 +1330,7 @@ impl DataUsageCache {
             self.info.scan_checkpoint = None;
             self.info.scan_raw_enumeration_cursor = None;
             self.info.scan_raw_enumeration_page_index = None;
+            self.info.scan_raw_enumeration_path_page_indexes.clear();
             self.info.scan_coverage_receipt = None;
         }
         // Old readers do not understand coverage sweeps. An absent plan makes
@@ -1230,22 +1474,37 @@ impl DataUsageCache {
     }
 
     pub(crate) fn validated_raw_enumeration_page_index(&self) -> Option<&RawEnumerationPageIndex> {
-        let index = self.info.scan_raw_enumeration_page_index.as_ref()?;
+        self.info
+            .scan_raw_enumeration_page_index
+            .as_ref()
+            .filter(|index| self.raw_enumeration_page_index_is_valid(index))
+    }
+
+    /// The durable index of one directory on the saved walk path.
+    pub(crate) fn validated_raw_enumeration_page_index_for_parent(&self, parent: &str) -> Option<&RawEnumerationPageIndex> {
+        self.info
+            .scan_raw_enumeration_page_index
+            .iter()
+            .chain(&self.info.scan_raw_enumeration_path_page_indexes)
+            .find(|index| index.parent() == Some(parent))
+            .filter(|index| self.raw_enumeration_page_index_is_valid(index))
+    }
+
+    fn raw_enumeration_page_index_is_valid(&self, index: &RawEnumerationPageIndex) -> bool {
         if self.info.scan_progress.is_none()
             || !self.info.scan_identity.is_some_and(|identity| identity.is_valid())
             || self.info.source.is_none()
             || index.committed_entries().is_err()
             || index.indexed_entries().is_err()
         {
-            return None;
+            return false;
         }
-        let parent = match index.status() {
-            RawEnumerationPageOwnerStatus::Unsupported => return None,
+        match index.status() {
+            RawEnumerationPageOwnerStatus::Unsupported => false,
             RawEnumerationPageOwnerStatus::Building { parent, .. } | RawEnumerationPageOwnerStatus::Ready { parent, .. } => {
-                parent
+                path_is_in_bucket_scope(&self.info.name, &parent)
             }
-        };
-        path_is_in_bucket_scope(&self.info.name, &parent).then_some(index)
+        }
     }
 
     /// Seal only the frontier supplied by completed traversal, never a restored cursor.

crates/scanner/src/data_usage_define/tests.rs

@@ -57,6 +57,66 @@ fn scoped_scan_coverage_metadata_preserves_map_compatibility() {
     assert_eq!(decoded.scan_coverage_digest, Some(TEST_PLAN_DIGEST));
 }
 
+#[test]
+fn raw_enumeration_path_indexes_preserve_map_compatibility_and_validation() {
+    #[derive(serde::Deserialize)]
+    struct LegacyInfo {
+        name: String,
+        scan_raw_enumeration_page_index: Option<RawEnumerationPageIndex>,
+    }
+    let index_for = |parent: &str| RawEnumerationPageIndex::new(parent, 8).expect("fixture index");
+    let mut cache = DataUsageCache {
+        info: DataUsageCacheInfo {
+            name: "bucket".to_string(),
+            scan_progress: Some(DataUsageScanProgress {
+                started_plan: TEST_PLAN_DIGEST,
+                requested_plan: TEST_PLAN_DIGEST,
+            }),
+            scan_identity: Some(DataUsageScanIdentity {
+                version: 1,
+                bucket_incarnation: uuid::Uuid::from_u128(1),
+                set_layout: DataUsageScanPlanDigest([2; 32]),
+                publication_epoch: 1,
+                tier_registry_generation: 0,
+                scan_mode: HealScanMode::Normal,
+            }),
+            source: Some(DataUsageCacheSource::new(0, 0)),
+            scan_raw_enumeration_page_index: Some(index_for("bucket/wide")),
+            scan_raw_enumeration_path_page_indexes: vec![index_for("bucket"), index_for("other-bucket/wide")],
+            ..Default::default()
+        },
+        cache: HashMap::new(),
+    };
+    assert!(cache.validated_raw_enumeration_page_index_for_parent("bucket").is_some());
+    assert!(
+        cache
+            .validated_raw_enumeration_page_index_for_parent("other-bucket/wide")
+            .is_none(),
+        "a saved path index must stay inside its bucket"
+    );
+
+    let encoded = rmp_serde::to_vec(&cache.info).expect("path indexes should remain map encoded");
+    let legacy: LegacyInfo = rmp_serde::from_slice(&encoded).expect("old map readers should ignore the path indexes");
+    assert_eq!(legacy.name, "bucket");
+    assert_eq!(
+        legacy
+            .scan_raw_enumeration_page_index
+            .and_then(|index| index.parent().map(str::to_owned)),
+        Some("bucket/wide".to_string())
+    );
+    let decoded: DataUsageCacheInfo = rmp_serde::from_slice(&encoded).expect("new reader should restore the path indexes");
+    assert_eq!(decoded.scan_raw_enumeration_path_page_indexes.len(), 2);
+
+    cache.info.scan_raw_enumeration_path_page_indexes.clear();
+    let encoded = rmp_serde::to_vec(&cache.info).expect("empty path indexes should encode");
+    assert!(
+        !encoded
+            .windows(b"scan_raw_enumeration_path_page_indexes".len())
+            .any(|window| window == b"scan_raw_enumeration_path_page_indexes"),
+        "the default must stay absent from the map"
+    );
+}
+
 #[derive(Debug, PartialEq, Eq)]
 struct CachePutRecord {
     object: String,

crates/scanner/src/remote_scanner/stream.rs

@@ -80,6 +80,7 @@ impl RemoteScannerBudget {
             max_duration: self.max_duration_ms.map(Duration::from_millis),
             max_objects: self.max_objects,
             max_directories: self.max_directories,
+            ..Default::default()
         }
     }
 }

crates/scanner/src/runtime_config.rs

@@ -743,6 +743,7 @@ pub(crate) fn lookup_scanner_runtime_config(
         max_duration: cycle_max_duration.filter(|duration| !duration.is_zero()),
         max_objects: cycle_max_objects,
         max_directories: cycle_max_directories,
+        ..Default::default()
     };
 
     let (bitrot_cycle, bitrot_cycle_source) = if let Some(value) = rustfs_utils::get_env_opt_str(ENV_SCANNER_BITROT_CYCLE_SECS) {
@@ -1171,6 +1172,7 @@ mod tests {
                 max_duration: Some(Duration::from_secs(1800)),
                 max_objects: Some(1000),
                 max_directories: Some(8000),
+                ..Default::default()
             },
             ..Default::default()
         };

crates/scanner/src/scanner.rs

@@ -474,14 +474,17 @@ fn scan_cycle_partial_reason(reason: Option<ScannerCycleBudgetReason>) -> ScanCy
     match reason {
         Some(ScannerCycleBudgetReason::Runtime) => ScanCyclePartialReason::Runtime,
         Some(ScannerCycleBudgetReason::Objects) => ScanCyclePartialReason::Objects,
-        Some(ScannerCycleBudgetReason::Directories) => ScanCyclePartialReason::Directories,
+        // Raw enumeration is directory work; the reason set is a stable metric label.
+        Some(ScannerCycleBudgetReason::Directories | ScannerCycleBudgetReason::Entries) => ScanCyclePartialReason::Directories,
         None => ScanCyclePartialReason::Unknown,
     }
 }
 
 fn scan_cycle_partial_source(reason: Option<ScannerCycleBudgetReason>) -> Option<ScannerWorkSource> {
     match reason {
-        Some(ScannerCycleBudgetReason::Objects | ScannerCycleBudgetReason::Directories) => Some(ScannerWorkSource::Usage),
+        Some(ScannerCycleBudgetReason::Objects | ScannerCycleBudgetReason::Directories | ScannerCycleBudgetReason::Entries) => {
+            Some(ScannerWorkSource::Usage)
+        }
         Some(ScannerCycleBudgetReason::Runtime) | None => None,
     }
 }

crates/scanner/src/scanner_budget.rs

@@ -24,13 +24,51 @@ 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 BUDGET_REASON_ENTRIES: u8 = 4;
 const PROGRESS_CLOCK_SAMPLE_INTERVAL: u64 = 128;
+/// One raw-enumeration page, so a directory stop always commits durable work.
+const DIRECTORY_ENTRY_BUDGET_FLOOR: u64 = 128;
+/// Directory entries one object or directory budget unit pays for. A forward
+/// sweep classifies in sorted order and so needs a whole listing first; a
+/// directory a few times wider than the budget must still reach objects in
+/// its first round, while a far wider one builds its index across rounds.
+const DIRECTORY_ENTRIES_PER_COUNT_UNIT: u64 = 16;
 
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub(crate) struct ScannerCycleBudgetConfig {
     pub max_duration: Option<Duration>,
     pub max_objects: Option<u64>,
     pub max_directories: Option<u64>,
+    /// Raw directory entries first observed under this budget. Entries replayed
+    /// from a validated page index are not charged.
+    pub max_entries: Option<u64>,
+    /// Cumulative name bytes of the charged raw entries.
+    pub max_entry_name_bytes: Option<u64>,
+    /// One bucket's share. `None` lets a bucket consume the whole budget.
+    pub bucket_quantum: Option<ScannerBucketQuantum>,
+}
+
+/// Limits of one bucket admission, charged to the owning cycle budget as well.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub(crate) struct ScannerBucketQuantum {
+    pub max_duration: Option<Duration>,
+    pub max_objects: Option<u64>,
+    pub max_directories: Option<u64>,
+    pub max_entries: Option<u64>,
+    pub max_entry_name_bytes: Option<u64>,
+}
+
+impl From<ScannerBucketQuantum> for ScannerCycleBudgetConfig {
+    fn from(quantum: ScannerBucketQuantum) -> Self {
+        Self {
+            max_duration: quantum.max_duration,
+            max_objects: quantum.max_objects,
+            max_directories: quantum.max_directories,
+            max_entries: quantum.max_entries,
+            max_entry_name_bytes: quantum.max_entry_name_bytes,
+            bucket_quantum: None,
+        }
+    }
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -38,6 +76,7 @@ pub(crate) enum ScannerCycleBudgetReason {
     Runtime,
     Objects,
     Directories,
+    Entries,
 }
 
 impl ScannerCycleBudgetReason {
@@ -46,6 +85,7 @@ impl ScannerCycleBudgetReason {
             Self::Runtime => BUDGET_REASON_RUNTIME,
             Self::Objects => BUDGET_REASON_OBJECTS,
             Self::Directories => BUDGET_REASON_DIRECTORIES,
+            Self::Entries => BUDGET_REASON_ENTRIES,
         }
     }
 
@@ -54,12 +94,15 @@ impl ScannerCycleBudgetReason {
             BUDGET_REASON_RUNTIME => Some(Self::Runtime),
             BUDGET_REASON_OBJECTS => Some(Self::Objects),
             BUDGET_REASON_DIRECTORIES => Some(Self::Directories),
+            BUDGET_REASON_ENTRIES => Some(Self::Entries),
             _ => None,
         }
     }
 }
 
 pub struct ScannerCycleBudget {
+    /// The cycle budget a bucket quantum also charges.
+    parent: Option<Arc<ScannerCycleBudget>>,
     token: CancellationToken,
     reason: Arc<AtomicU8>,
     started_at: Instant,
@@ -67,11 +110,17 @@ pub struct ScannerCycleBudget {
     max_duration: Option<Duration>,
     max_objects: Option<u64>,
     max_directories: Option<u64>,
+    max_entries: Option<u64>,
+    max_entry_name_bytes: Option<u64>,
+    directory_entry_limit: Option<u64>,
+    bucket_quantum: Option<ScannerBucketQuantum>,
     track_progress: bool,
     track_unbounded_counts: bool,
     objects_scanned: AtomicU64,
     directories_started: AtomicU64,
     entries_visited: AtomicU64,
+    entries_charged: AtomicU64,
+    entry_name_bytes_charged: AtomicU64,
     last_progress_millis: AtomicU64,
     cycle_state_persisted: AtomicBool,
     remote_scan_progress: Arc<RemoteScanProgress>,
@@ -80,16 +129,16 @@ pub struct ScannerCycleBudget {
 impl ScannerCycleBudget {
     #[cfg(test)]
     pub(crate) fn new(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc<Self> {
-        Self::new_inner(parent, config, false, false)
+        Self::new_inner(parent, config, false, false, None)
     }
 
     pub(crate) fn new_with_progress_tracking(parent: &CancellationToken, config: ScannerCycleBudgetConfig) -> Arc<Self> {
-        Self::new_inner(parent, config, true, true)
+        Self::new_inner(parent, config, true, true, None)
     }
 
     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)
+        Self::new_inner(parent, config, track_progress, false, None)
     }
 
     fn new_inner(
@@ -97,6 +146,7 @@ impl ScannerCycleBudget {
         config: ScannerCycleBudgetConfig,
         track_progress: bool,
         track_unbounded_counts: bool,
+        cycle_budget: Option<Arc<ScannerCycleBudget>>,
     ) -> Arc<Self> {
         let token = parent.child_token();
         let reason = Arc::new(AtomicU8::new(BUDGET_REASON_NONE));
@@ -124,7 +174,11 @@ impl ScannerCycleBudget {
             });
         }
 
+        let remote_scan_progress = cycle_budget
+            .as_ref()
+            .map_or_else(|| Arc::new(RemoteScanProgress::default()), |budget| budget.remote_scan_progress());
         Arc::new(Self {
+            parent: cycle_budget,
             token,
             reason,
             started_at,
@@ -132,14 +186,20 @@ impl ScannerCycleBudget {
             max_duration: config.max_duration,
             max_objects: config.max_objects,
             max_directories: config.max_directories,
+            max_entries: config.max_entries,
+            max_entry_name_bytes: config.max_entry_name_bytes,
+            directory_entry_limit: directory_entry_limit(&config),
+            bucket_quantum: config.bucket_quantum,
             track_progress,
             track_unbounded_counts,
             objects_scanned: AtomicU64::new(0),
             directories_started: AtomicU64::new(0),
             entries_visited: AtomicU64::new(0),
+            entries_charged: AtomicU64::new(0),
+            entry_name_bytes_charged: AtomicU64::new(0),
             last_progress_millis: AtomicU64::new(0),
             cycle_state_persisted: AtomicBool::new(false),
-            remote_scan_progress: Arc::new(RemoteScanProgress::default()),
+            remote_scan_progress,
         })
     }
 
@@ -147,12 +207,31 @@ impl ScannerCycleBudget {
         self.token.clone()
     }
 
+    /// Budget and cancellation scope of one bucket scan. Without a quantum the
+    /// bucket shares this budget; a quantum stops only its own scan token.
+    pub(crate) fn for_bucket(self: &Arc<Self>, ctx: &CancellationToken) -> (Arc<Self>, CancellationToken) {
+        let Some(quantum) = self.bucket_quantum else {
+            return (self.clone(), ctx.child_token());
+        };
+        let budget = Self::new_inner(ctx, quantum.into(), self.track_progress, self.track_unbounded_counts, Some(self.clone()));
+        let token = budget.token();
+        (budget, token)
+    }
+
+    /// True when only this bucket's share ended and the cycle can continue.
+    pub(crate) fn quantum_exhausted(&self) -> bool {
+        self.parent.as_ref().is_some_and(|parent| !parent.budget_elapsed())
+            && self.reason.load(Ordering::Relaxed) != BUDGET_REASON_NONE
+    }
+
     pub(crate) fn budget_elapsed(&self) -> bool {
         self.reason.load(Ordering::Relaxed) != BUDGET_REASON_NONE
+            || self.parent.as_ref().is_some_and(|parent| parent.budget_elapsed())
     }
 
     pub(crate) fn reason(&self) -> Option<ScannerCycleBudgetReason> {
         ScannerCycleBudgetReason::from_code(self.reason.load(Ordering::Relaxed))
+            .or_else(|| self.parent.as_ref().and_then(|parent| parent.reason()))
     }
 
     pub(crate) fn max_duration(&self) -> Option<Duration> {
@@ -176,7 +255,10 @@ impl ScannerCycleBudget {
     }
 
     pub(crate) fn requires_serial_progress_accounting(&self) -> bool {
-        self.max_objects.is_some() || self.max_directories.is_some()
+        self.max_objects.is_some()
+            || self.max_directories.is_some()
+            || self.max_entries.is_some()
+            || self.max_entry_name_bytes.is_some()
     }
 
     pub(crate) fn remaining_config(&self) -> ScannerCycleBudgetConfig {
@@ -195,6 +277,13 @@ impl ScannerCycleBudget {
             max_directories: self
                 .max_directories
                 .map(|max| max.saturating_sub(self.directories_started.load(Ordering::Relaxed))),
+            max_entries: self
+                .max_entries
+                .map(|max| max.saturating_sub(self.entries_charged.load(Ordering::Relaxed))),
+            max_entry_name_bytes: self
+                .max_entry_name_bytes
+                .map(|max| max.saturating_sub(self.entry_name_bytes_charged.load(Ordering::Relaxed))),
+            bucket_quantum: None,
         }
     }
 
@@ -234,6 +323,9 @@ impl ScannerCycleBudget {
     }
 
     pub(crate) fn record_entries_visited(&self, entries_visited: u64) {
+        if let Some(parent) = &self.parent {
+            parent.record_entries_visited(entries_visited);
+        }
         if self.track_progress {
             let entries = saturating_fetch_add(&self.entries_visited, entries_visited);
             self.record_progress_sample(entries);
@@ -245,7 +337,11 @@ impl ScannerCycleBudget {
     }
 
     pub(crate) fn record_remote_progress(&self, objects_scanned: u64, directories_started: u64) {
-        self.remote_scan_progress.record(objects_scanned, directories_started);
+        // A quantum shares the cycle's progress record; its parent writes it.
+        match &self.parent {
+            Some(parent) => parent.record_remote_progress(objects_scanned, directories_started),
+            None => self.remote_scan_progress.record(objects_scanned, directories_started),
+        }
         if self.track_progress || self.max_objects.is_some() {
             let objects = saturating_fetch_add(&self.objects_scanned, objects_scanned);
             if self.track_progress {
@@ -271,6 +367,9 @@ impl ScannerCycleBudget {
     }
 
     pub(crate) fn cancel_after_unreported_remote_progress(&self) {
+        if let Some(parent) = &self.parent {
+            parent.cancel_after_unreported_remote_progress();
+        }
         if self.max_objects.is_some() {
             self.cancel_for(ScannerCycleBudgetReason::Objects);
         } else if self.max_directories.is_some() {
@@ -279,6 +378,11 @@ impl ScannerCycleBudget {
     }
 
     pub(crate) fn try_start_directory(&self) -> bool {
+        let parent_allows = self.parent.as_ref().is_none_or(|parent| parent.try_start_directory());
+        self.try_start_own_directory() && parent_allows
+    }
+
+    fn try_start_own_directory(&self) -> bool {
         if self.max_directories.is_none() && !self.track_unbounded_counts {
             return true;
         }
@@ -298,7 +402,35 @@ impl ScannerCycleBudget {
         true
     }
 
+    /// Charges one raw entry first observed in this scan. `directory_entries`
+    /// counts the charged entries of the current directory: a count budget is
+    /// a total work limit, so one listing cannot read past it.
+    ///
+    /// Returns `false` after charging the entry that exhausts a limit; the
+    /// caller stops with the entry already recorded in its durable index.
+    pub(crate) fn try_charge_raw_entry(&self, name_bytes: u64, directory_entries: u64) -> bool {
+        let parent_allows = self
+            .parent
+            .as_ref()
+            .is_none_or(|parent| parent.try_charge_raw_entry(name_bytes, directory_entries));
+        let entries = self
+            .max_entries
+            .is_some_and(|max| saturating_fetch_add(&self.entries_charged, 1) >= max);
+        let name_bytes = self
+            .max_entry_name_bytes
+            .is_some_and(|max| saturating_fetch_add(&self.entry_name_bytes_charged, name_bytes) >= max);
+        let directory = self.directory_entry_limit.is_some_and(|limit| directory_entries >= limit);
+        if entries || name_bytes || directory {
+            self.cancel_for(ScannerCycleBudgetReason::Entries);
+            return false;
+        }
+        parent_allows
+    }
+
     pub(crate) fn record_object_scanned(&self) {
+        if let Some(parent) = &self.parent {
+            parent.record_object_scanned();
+        }
         if self.max_objects.is_none() && !self.track_unbounded_counts {
             return;
         }
@@ -337,6 +469,21 @@ fn saturating_fetch_add(value: &AtomicU64, delta: u64) -> u64 {
     }
 }
 
+fn directory_entry_limit(config: &ScannerCycleBudgetConfig) -> Option<u64> {
+    if config.max_objects.is_none() && config.max_directories.is_none() {
+        return None;
+    }
+    let total = config
+        .max_objects
+        .unwrap_or(0)
+        .saturating_add(config.max_directories.unwrap_or(0));
+    Some(
+        total
+            .saturating_mul(DIRECTORY_ENTRIES_PER_COUNT_UNIT)
+            .max(DIRECTORY_ENTRY_BUDGET_FLOOR),
+    )
+}
+
 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
@@ -354,6 +501,127 @@ impl Drop for ScannerCycleBudget {
 mod tests {
     use super::*;
 
+    #[tokio::test]
+    async fn fair_prefix_quantum_budget_limits_entries_and_name_bytes_with_one_reason() {
+        let parent = CancellationToken::new();
+        let by_entries = ScannerCycleBudget::new(
+            &parent,
+            ScannerCycleBudgetConfig {
+                max_entries: Some(2),
+                ..Default::default()
+            },
+        );
+        assert!(by_entries.try_charge_raw_entry(1, 1));
+        assert!(!by_entries.try_charge_raw_entry(1, 2));
+        assert_eq!(by_entries.reason(), Some(ScannerCycleBudgetReason::Entries));
+        assert!(by_entries.token().is_cancelled());
+
+        let by_bytes = ScannerCycleBudget::new(
+            &parent,
+            ScannerCycleBudgetConfig {
+                max_entry_name_bytes: Some(10),
+                ..Default::default()
+            },
+        );
+        assert!(by_bytes.try_charge_raw_entry(9, 1));
+        assert!(!by_bytes.try_charge_raw_entry(1, 2));
+        assert_eq!(by_bytes.reason(), Some(ScannerCycleBudgetReason::Entries));
+        assert!(!parent.is_cancelled(), "a budget stop must not cancel its parent token");
+    }
+
+    #[tokio::test]
+    async fn fair_prefix_quantum_budget_derives_a_directory_limit_only_from_count_budgets() {
+        let parent = CancellationToken::new();
+        let runtime_only = ScannerCycleBudget::new(&parent, ScannerCycleBudgetConfig::default());
+        assert!(runtime_only.try_charge_raw_entry(1, u64::MAX));
+
+        let small = ScannerCycleBudget::new(
+            &parent,
+            ScannerCycleBudgetConfig {
+                max_objects: Some(1),
+                ..Default::default()
+            },
+        );
+        assert!(small.try_charge_raw_entry(1, DIRECTORY_ENTRY_BUDGET_FLOOR - 1));
+        assert!(!small.try_charge_raw_entry(1, DIRECTORY_ENTRY_BUDGET_FLOOR));
+
+        let large = ScannerCycleBudget::new(
+            &parent,
+            ScannerCycleBudgetConfig {
+                max_objects: Some(1000),
+                max_directories: Some(24),
+                ..Default::default()
+            },
+        );
+        assert!(large.try_charge_raw_entry(1, 1024 * DIRECTORY_ENTRIES_PER_COUNT_UNIT - 1));
+        assert!(!large.try_charge_raw_entry(1, 1024 * DIRECTORY_ENTRIES_PER_COUNT_UNIT));
+    }
+
+    #[tokio::test]
+    async fn fair_prefix_quantum_bucket_share_stops_only_its_own_scan() {
+        let ctx = CancellationToken::new();
+        let cycle = ScannerCycleBudget::new_with_progress_tracking(
+            &ctx,
+            ScannerCycleBudgetConfig {
+                max_objects: Some(3),
+                bucket_quantum: Some(ScannerBucketQuantum {
+                    max_objects: Some(2),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            },
+        );
+        let (first, first_token) = cycle.for_bucket(&cycle.token());
+        first.record_object_scanned();
+        first.record_object_scanned();
+        assert!(first.quantum_exhausted());
+        assert!(first_token.is_cancelled());
+        assert_eq!(first.reason(), Some(ScannerCycleBudgetReason::Objects));
+        assert!(!cycle.budget_elapsed(), "a bucket share leaves the cycle running");
+        assert!(!cycle.token().is_cancelled());
+        assert_eq!(cycle.progress().0, 2, "bucket work is charged to the cycle");
+
+        let (second, second_token) = cycle.for_bucket(&cycle.token());
+        assert!(!second.budget_elapsed());
+        second.record_object_scanned();
+        assert!(cycle.budget_elapsed(), "the total budget still bounds every share");
+        assert!(second_token.is_cancelled());
+        assert!(second.budget_elapsed());
+        assert!(!second.quantum_exhausted(), "a cycle stop is not a quantum handoff");
+    }
+
+    #[tokio::test]
+    async fn fair_prefix_quantum_bucket_share_runtime_expires_independently() {
+        let ctx = CancellationToken::new();
+        let cycle = ScannerCycleBudget::new(
+            &ctx,
+            ScannerCycleBudgetConfig {
+                bucket_quantum: Some(ScannerBucketQuantum {
+                    max_duration: Some(Duration::ZERO),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            },
+        );
+        let (bucket, token) = cycle.for_bucket(&cycle.token());
+        tokio::time::timeout(Duration::from_secs(5), token.cancelled())
+            .await
+            .expect("quantum deadline should cancel the bucket scan");
+        assert_eq!(bucket.reason(), Some(ScannerCycleBudgetReason::Runtime));
+        assert!(!cycle.budget_elapsed());
+    }
+
+    #[tokio::test]
+    async fn bucket_without_a_quantum_shares_the_cycle_budget() {
+        let ctx = CancellationToken::new();
+        let cycle = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default());
+        let (bucket, token) = cycle.for_bucket(&cycle.token());
+        assert!(Arc::ptr_eq(&bucket, &cycle));
+        cycle.cancel_for_runtime();
+        assert!(token.is_cancelled());
+        assert!(!bucket.quantum_exhausted());
+    }
+
     #[tokio::test]
     async fn runtime_budget_cancels_child_token() {
         let parent = CancellationToken::new();
@@ -441,6 +709,7 @@ mod tests {
                 max_duration: Some(Duration::from_secs(60)),
                 max_objects: Some(10),
                 max_directories: Some(5),
+                ..Default::default()
             },
         );
 

crates/scanner/src/scanner_folder.rs

@@ -498,7 +498,10 @@ fn checkpoint_reason_from_budget(reason: Option<ScannerCycleBudgetReason>) -> Da
     match reason {
         Some(ScannerCycleBudgetReason::Runtime) => DataUsageScanCheckpointReason::Runtime,
         Some(ScannerCycleBudgetReason::Objects) => DataUsageScanCheckpointReason::Objects,
-        Some(ScannerCycleBudgetReason::Directories) => DataUsageScanCheckpointReason::Directories,
+        // This enum is persisted and older readers reject unknown variants.
+        Some(ScannerCycleBudgetReason::Directories | ScannerCycleBudgetReason::Entries) => {
+            DataUsageScanCheckpointReason::Directories
+        }
         None => DataUsageScanCheckpointReason::Unknown,
     }
 }
@@ -1302,9 +1305,11 @@ impl FolderScanner {
         }
     }
 
-    fn record_raw_enumeration_entry(&mut self, parent: &str, entry: &str) {
+    /// Records one raw entry. Returns whether a durable index holds it, which
+    /// is what makes stopping at this entry recoverable.
+    fn record_raw_enumeration_entry(&mut self, parent: &str, entry: &str) -> bool {
         if self.old_cache.info.scan_progress.is_none() {
-            return;
+            return false;
         }
         if let Some(position) = self
             .raw_enumeration_progress
@@ -1317,18 +1322,14 @@ impl FolderScanner {
             self.raw_enumeration_progress
                 .push(RawEnumerationProgress::new(parent, page_index));
         }
-        if let Some(progress) = self.raw_enumeration_progress.last_mut() {
+        self.raw_enumeration_progress.last_mut().is_some_and(|progress| {
             progress.record_entry(entry);
-        }
+            progress.page_index.is_some()
+        })
     }
 
     fn raw_enumeration_page_index_for_parent(&self, parent: &str) -> Option<&RawEnumerationPageIndex> {
-        self.old_cache
-            .info
-            .scan_raw_enumeration_page_index
-            .as_ref()
-            .filter(|index| index.parent() == Some(parent))?;
-        self.old_cache.validated_raw_enumeration_page_index()
+        self.old_cache.validated_raw_enumeration_page_index_for_parent(parent)
     }
 
     fn raw_enumeration_committed_entry_oracle(&self, parent: &str) -> HashSet<String> {
@@ -1372,23 +1373,36 @@ impl FolderScanner {
         if self.raw_enumeration_progress.is_empty() {
             return (None, None);
         }
-        let progress_index = self
-            .raw_enumeration_progress
-            .iter()
-            .enumerate()
-            .max_by_key(|(index, progress)| (progress.checkpointable_entry_count(), std::cmp::Reverse(*index)))
-            .map(|(index, _)| index)
-            .unwrap_or(0);
+        let progress_index = self.selected_raw_enumeration_progress().unwrap_or(0);
         let progress = self.raw_enumeration_progress.swap_remove(progress_index);
         self.raw_enumeration_progress.clear();
         (progress.cursor(), progress.page_index())
     }
 
-    fn peek_raw_enumeration_resume_state(&self) -> (Option<DataUsageRawEnumerationCursor>, Option<RawEnumerationPageIndex>) {
+    fn selected_raw_enumeration_progress(&self) -> Option<usize> {
         self.raw_enumeration_progress
             .iter()
-            .max_by_key(|progress| progress.checkpointable_entry_count())
-            .map_or((None, None), |progress| (progress.cursor(), progress.page_index()))
+            .enumerate()
+            .max_by_key(|(index, progress)| (progress.checkpointable_entry_count(), std::cmp::Reverse(*index)))
+            .map(|(index, _)| index)
+    }
+
+    /// The walk path except the directory the resume state selects.
+    fn raw_enumeration_path_page_indexes(&self) -> Vec<RawEnumerationPageIndex> {
+        let selected = self.selected_raw_enumeration_progress();
+        self.raw_enumeration_progress
+            .iter()
+            .enumerate()
+            .filter(|(index, _)| Some(*index) != selected)
+            .filter_map(|(_, progress)| progress.page_index())
+            .collect()
+    }
+
+    fn peek_raw_enumeration_resume_state(&self) -> (Option<DataUsageRawEnumerationCursor>, Option<RawEnumerationPageIndex>) {
+        self.selected_raw_enumeration_progress().map_or((None, None), |index| {
+            let progress = &self.raw_enumeration_progress[index];
+            (progress.cursor(), progress.page_index())
+        })
     }
 
     async fn scan_child_folder(
@@ -1480,6 +1494,7 @@ impl FolderScanner {
         snapshot.info.last_update = Some(SystemTime::now());
         snapshot.info.snapshot_complete = false;
         let (cursor, page_index) = self.peek_raw_enumeration_resume_state();
+        snapshot.info.scan_raw_enumeration_path_page_indexes = self.raw_enumeration_path_page_indexes();
         snapshot.info.scan_raw_enumeration_cursor = cursor;
         snapshot.info.scan_raw_enumeration_page_index = page_index;
         snapshot.seal_scan_frontier(self.coverage_frontier.as_deref()).ok()?;
@@ -1976,6 +1991,9 @@ impl FolderScanner {
             let mut last_entry_progress = Instant::now();
             let mut raw_enumeration_complete = false;
             let raw_enumeration_committed_entries = self.raw_enumeration_committed_entry_oracle(&folder.name);
+            let raw_replay_floor = u64::try_from(raw_enumeration_committed_entries.len()).unwrap_or(u64::MAX);
+            let mut raw_entries_observed = 0_u64;
+            let mut raw_entries_charged = 0_u64;
 
             let mut candidate_unclassified = mode != FolderScanMode::Namespace;
             let mut deferred_metadata = None;
@@ -2061,8 +2079,13 @@ impl FolderScanner {
                     continue;
                 }
                 if !replayed_metadata {
-                    let raw_entry_consumed_by_owner_index = raw_enumeration_committed_entries.contains(&file_name);
-                    if !raw_entry_consumed_by_owner_index {
+                    // A restored index appends one pending name per entry past
+                    // its old size, so charging only that tail keeps each
+                    // budget unit durable even if the listing order changed.
+                    raw_entries_observed = raw_entries_observed.saturating_add(1);
+                    let raw_entry_is_new =
+                        raw_entries_observed > raw_replay_floor && !raw_enumeration_committed_entries.contains(&file_name);
+                    if raw_entry_is_new {
                         #[cfg(test)]
                         tests::enumeration_restart::observe_raw_entry(&dir_path, &entry.file_name(), &self.budget);
                         pending_entry_progress = pending_entry_progress.saturating_add(1);
@@ -2074,7 +2097,18 @@ impl FolderScanner {
                             last_entry_progress = Instant::now();
                         }
                     }
-                    self.record_raw_enumeration_entry(&folder.name, &file_name);
+                    let raw_entry_is_durable = self.record_raw_enumeration_entry(&folder.name, &file_name);
+                    // Object metadata and the entries behind it are charged to
+                    // the object budget; this one covers listing work.
+                    let object_entry = file_name == STORAGE_FORMAT_FILE || deferred_metadata.is_some();
+                    if raw_entry_is_new && raw_entry_is_durable && !object_entry {
+                        raw_entries_charged = raw_entries_charged.saturating_add(1);
+                        let name_bytes = u64::try_from(file_name.len()).unwrap_or(u64::MAX);
+                        if !self.budget.try_charge_raw_entry(name_bytes, raw_entries_charged) {
+                            self.budget.record_entries_visited(pending_entry_progress);
+                            return Err(ScannerError::Other("Operation cancelled".to_string()));
+                        }
+                    }
                     self.maybe_send_checkpoint(&folder, &this_hash, into);
                 }
                 let is_storage_format_entry = file_name == STORAGE_FORMAT_FILE;
@@ -2474,7 +2508,9 @@ impl FolderScanner {
                 }
             }
             self.budget.record_entries_visited(pending_entry_progress);
-            if raw_enumeration_complete {
+            // A directory with queued children keeps its index until they end:
+            // a restart replays this listing to reach the unfinished child.
+            if raw_enumeration_complete && queued_folders.is_empty() {
                 self.finish_raw_enumeration_parent(&folder.name);
             }
 
@@ -2683,10 +2719,12 @@ impl FolderScanner {
                                 "Scanner child folder scan failed"
                             );
                             self.coverage_gap |= forward_sweep;
+                            self.finish_raw_enumeration_parent(&folder_item.name);
                             continue;
                         }
                     }
                 };
+                self.finish_raw_enumeration_parent(&folder_item.name);
                 if outcome == FolderScanOutcome::Vanished {
                     self.new_cache.delete_recursive(&h);
                     if let Some(updates) = &mut self.usage_updates {
@@ -3405,6 +3443,7 @@ pub(crate) async fn scan_data_folder_scoped(
             new_cache.info.scan_checkpoint = None;
             new_cache.info.scan_raw_enumeration_cursor = None;
             new_cache.info.scan_raw_enumeration_page_index = None;
+            new_cache.info.scan_raw_enumeration_path_page_indexes.clear();
             new_cache.info.scan_coverage_receipt = None;
             if had_scan_checkpoint {
                 global_metrics().record_scanner_checkpoint_cleared();
@@ -3443,6 +3482,7 @@ pub(crate) async fn scan_data_folder_scoped(
                 let root_hash = hash_path(&bucket_name);
                 let root_has_progress = data_usage_root_has_progress(&root);
                 let pending_heals_changed = scanner.pending_heals_changed;
+                let raw_enumeration_path_page_indexes = scanner.raw_enumeration_path_page_indexes();
                 let (raw_enumeration_cursor, raw_enumeration_page_index) = scanner.take_raw_enumeration_resume_state();
                 let carry_forward_cache = ((raw_enumeration_cursor.is_some() || raw_enumeration_page_index.is_some())
                     && !root_has_progress)
@@ -3465,6 +3505,7 @@ pub(crate) async fn scan_data_folder_scoped(
                 }
                 if raw_enumeration_page_index.is_some() {
                     new_cache.info.scan_raw_enumeration_page_index = raw_enumeration_page_index;
+                    new_cache.info.scan_raw_enumeration_path_page_indexes = raw_enumeration_path_page_indexes;
                     new_cache.info.scan_checkpoint = None;
                     new_cache.info.scan_resume_after = None;
                     new_cache.info.scan_coverage_receipt = None;
@@ -3495,6 +3536,7 @@ pub(crate) async fn scan_data_folder_scoped(
                             global_metrics().record_scanner_checkpoint_cleared();
                         }
                     }
+                    new_cache.info.scan_resource_deferred = matches!(&e, ScannerError::ResourceDeferred);
                     close_disk_guard.close().await;
                     return Err(ScannerError::PartialCache(Box::new(std::mem::take(new_cache))));
                 }
@@ -3524,6 +3566,7 @@ pub(crate) async fn scan_data_folder_scoped(
                         .insert(hash_path(&bucket_name).0, DataUsageEntry::default());
                 }
                 partial_cache.info.snapshot_complete = false;
+                partial_cache.info.scan_resource_deferred = true;
                 close_disk_guard.close().await;
                 return Err(ScannerError::PartialCache(Box::new(partial_cache)));
             }

crates/scanner/src/scanner_folder/tests.rs

@@ -30,6 +30,7 @@ use std::sync::Mutex;
 
 pub(super) mod checkpoint_fixture;
 pub(super) mod enumeration_restart;
+mod fair_prefix_quantum;
 mod incremental_enumeration;
 mod listing_completion;
 mod metadata_visits;

crates/scanner/src/scanner_folder/tests/enumeration_restart.rs

@@ -51,6 +51,10 @@ impl Drop for ObservationGuard {
     }
 }
 
+pub(in crate::scanner_folder) fn install_raw_entry_observer(root: PathBuf) -> ObservationGuard {
+    install_raw_entry_budget(root, u64::MAX)
+}
+
 pub(in crate::scanner_folder) fn install_raw_entry_budget(root: PathBuf, limit: u64) -> ObservationGuard {
     *OBSERVATION.lock().expect("install raw-entry observation") = Some(Observation {
         root,
@@ -83,6 +87,45 @@ pub(super) async fn read_bounded(path: &Path) -> Vec<u8> {
     bytes
 }
 
+/// Creates a flat synthetic bucket and the checkpointable cache a first sweep
+/// starts from.
+pub(super) async fn seed_flat_bucket(disk_root: &Path, objects: usize) -> DataUsageCache {
+    tokio::fs::create_dir(disk_root).await.expect("create fresh synthetic disk");
+    for index in 0..objects {
+        let object = format!("object-{index:04}");
+        let version = Uuid::from_u128(u128::try_from(index).expect("fixture index") + 1);
+        let bytes = metadata_for_object_version("bucket", &object, Some(version));
+        write_test_object_metadata_bytes(disk_root, "bucket", &object, &bytes).await;
+    }
+    let mut initial = DataUsageCache::default();
+    initial.info.name = "bucket".to_string();
+    let source = crate::data_usage_define::DataUsageCacheSource::new(0, 0);
+    let plan = crate::data_usage_define::DataUsageScanPlanDigest([31; 32]);
+    let identity = crate::data_usage_define::DataUsageScanIdentity {
+        version: 1,
+        bucket_incarnation: Uuid::from_u128(31),
+        set_layout: crate::data_usage_define::DataUsageScanPlanDigest([32; 32]),
+        publication_epoch: 1,
+        tier_registry_generation: 0,
+        scan_mode: HealScanMode::Normal,
+    };
+    assert_eq!(
+        initial.prepare_bucket_checkpoint("bucket", 1, 0, source, plan, identity),
+        crate::data_usage_define::DataUsageCachePrepareOutcome::Reset
+    );
+    initial.info.skip_healing = true;
+    initial.replace("bucket", "", DataUsageEntry::default());
+    initial
+}
+
+pub(super) fn observed_raw_entries() -> u64 {
+    OBSERVATION
+        .lock()
+        .expect("read raw-entry observation")
+        .as_ref()
+        .map_or(0, |observation| observation.entries)
+}
+
 async fn round(request: &Request) -> serde_json::Value {
     assert!((1..=1024).contains(&request.objects));
     assert!((1..=4096).contains(&request.raw_entry_budget));
@@ -90,31 +133,7 @@ async fn round(request: &Request) -> serde_json::Value {
     let disk_root = request.workspace.join("disk");
     let cache_path = request.workspace.join("cache.bin");
     if request.round == 0 {
-        tokio::fs::create_dir(&disk_root).await.expect("create fresh synthetic disk");
-        for index in 0..request.objects {
-            let object = format!("object-{index:04}");
-            let version = Uuid::from_u128(u128::try_from(index).expect("fixture index") + 1);
-            let bytes = metadata_for_object_version("bucket", &object, Some(version));
-            write_test_object_metadata_bytes(&disk_root, "bucket", &object, &bytes).await;
-        }
-        let mut initial = DataUsageCache::default();
-        initial.info.name = "bucket".to_string();
-        let source = crate::data_usage_define::DataUsageCacheSource::new(0, 0);
-        let plan = crate::data_usage_define::DataUsageScanPlanDigest([31; 32]);
-        let identity = crate::data_usage_define::DataUsageScanIdentity {
-            version: 1,
-            bucket_incarnation: Uuid::from_u128(31),
-            set_layout: crate::data_usage_define::DataUsageScanPlanDigest([32; 32]),
-            publication_epoch: 1,
-            tier_registry_generation: 0,
-            scan_mode: HealScanMode::Normal,
-        };
-        assert_eq!(
-            initial.prepare_bucket_checkpoint("bucket", 1, 0, source, plan, identity),
-            crate::data_usage_define::DataUsageCachePrepareOutcome::Reset
-        );
-        initial.info.skip_healing = true;
-        initial.replace("bucket", "", DataUsageEntry::default());
+        let initial = seed_flat_bucket(&disk_root, request.objects).await;
         tokio::fs::write(&cache_path, initial.marshal_msg().expect("initial cache codec"))
             .await
             .expect("persist initial cache");
@@ -137,10 +156,11 @@ async fn round(request: &Request) -> serde_json::Value {
         &parent,
         crate::scanner_budget::ScannerCycleBudgetConfig {
             max_objects: Some(request.raw_entry_budget),
+            max_entries: Some(request.raw_entry_budget),
             ..Default::default()
         },
     );
-    let _observation_guard = install_raw_entry_budget(disk.path(), request.raw_entry_budget);
+    let _observation_guard = install_raw_entry_observer(disk.path());
     let result = scan_data_folder(
         budget.token(),
         budget.clone(),

crates/scanner/src/scanner_folder/tests/fair_prefix_quantum.rs

@@ -0,0 +1,233 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::enumeration_restart::{install_raw_entry_observer, observed_raw_entries, seed_flat_bucket};
+use super::*;
+use crate::scanner_budget::ScannerCycleBudgetConfig;
+
+async fn open_fixture_disk(disk_root: &std::path::Path) -> Arc<Disk> {
+    let endpoint = Endpoint::try_from(disk_root.to_string_lossy().as_ref()).expect("fixture endpoint");
+    new_disk(
+        &endpoint,
+        &DiskOption {
+            cleanup: false,
+            health_check: false,
+        },
+    )
+    .await
+    .expect("open synthetic disk")
+}
+
+/// A count budget is the operator's total work limit. It must also stop the
+/// first listing of a directory far wider than that budget, instead of letting
+/// one read_dir pass finish before the first object is charged.
+#[tokio::test]
+#[serial]
+async fn fair_prefix_quantum_count_budget_bounds_the_first_directory_build() {
+    const OBJECTS: usize = 300;
+    const FIRST_BUILD_FLOOR: u64 = 128;
+    let temp = tempfile::tempdir().expect("fixture directory");
+    let disk_root = temp.path().join("disk");
+    let cache = seed_flat_bucket(&disk_root, OBJECTS).await;
+    let disk = open_fixture_disk(&disk_root).await;
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(
+        &parent,
+        ScannerCycleBudgetConfig {
+            max_objects: Some(8),
+            ..Default::default()
+        },
+    );
+    let _observation = install_raw_entry_observer(disk.path());
+
+    let result = scan_data_folder(
+        budget.token(),
+        budget.clone(),
+        vec![disk.clone()],
+        disk,
+        cache,
+        None,
+        HealScanMode::Normal,
+        SCANNER_SLEEPER.clone(),
+    )
+    .await;
+
+    assert!(
+        matches!(result, Err(ScannerError::PartialCache(_))),
+        "an exhausted first build must return recoverable progress"
+    );
+    assert!(
+        observed_raw_entries() <= FIRST_BUILD_FLOOR,
+        "first build read {} raw entries under an 8-object budget",
+        observed_raw_entries()
+    );
+}
+
+struct RestartRound {
+    outcome: DataUsageCache,
+    finished: bool,
+    observed_entries: u64,
+}
+
+/// One scanner lifetime: the input was decoded from persisted bytes, and the
+/// result is encoded and decoded again before the next round may use it.
+async fn scan_one_restart_round(
+    disk_root: &std::path::Path,
+    cache: DataUsageCache,
+    config: ScannerCycleBudgetConfig,
+) -> RestartRound {
+    let disk = open_fixture_disk(disk_root).await;
+    let parent = CancellationToken::new();
+    let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, config);
+    let _observation = install_raw_entry_observer(disk.path());
+    let result = scan_data_folder(
+        budget.token(),
+        budget.clone(),
+        vec![disk.clone()],
+        disk,
+        cache.clone(),
+        None,
+        HealScanMode::Normal,
+        SCANNER_SLEEPER.clone(),
+    )
+    .await;
+    let observed_entries = observed_raw_entries();
+    let (returned, finished) = match result {
+        Ok(cache) => (cache, true),
+        Err(ScannerError::PartialCache(cache)) => (*cache, false),
+        Err(error) => panic!("a budget stop must keep recoverable progress: {error}"),
+    };
+    let encoded = returned.marshal_msg().expect("encode returned cache");
+    RestartRound {
+        outcome: DataUsageCache::unmarshal(&encoded).expect("decode returned cache"),
+        finished,
+        observed_entries,
+    }
+}
+
+async fn run_until_finished(
+    disk_root: &std::path::Path,
+    mut cache: DataUsageCache,
+    config: ScannerCycleBudgetConfig,
+    rounds: usize,
+) -> DataUsageCache {
+    for round in 0..rounds {
+        let result = scan_one_restart_round(disk_root, cache, config).await;
+        if let Some(limit) = config.max_entries {
+            assert!(result.observed_entries <= limit, "round {round} read {} entries", result.observed_entries);
+        }
+        cache = result.outcome;
+        if result.finished {
+            assert!(cache.info.snapshot_complete, "a finished sweep must certify its bucket");
+            return cache;
+        }
+    }
+    panic!("the sweep did not finish in {rounds} restarts");
+}
+
+/// Both a wide bucket root and a wide nested prefix must keep their indexes:
+/// losing either one recharges its replay and a fixed budget stops advancing.
+#[tokio::test]
+#[serial]
+async fn fair_prefix_quantum_nested_wide_directories_converge_across_restarts() {
+    const WIDTH: usize = 40;
+    let temp = tempfile::tempdir().expect("fixture directory");
+    let disk_root = temp.path().join("disk");
+    let cache = seed_flat_bucket(&disk_root, WIDTH).await;
+    for index in 0..WIDTH {
+        let object = format!("wide/nested-{index:04}");
+        let version = Uuid::from_u128(u128::try_from(index).expect("fixture index") + 1000);
+        let bytes = metadata_for_object_version("bucket", &object, Some(version));
+        write_test_object_metadata_bytes(&disk_root, "bucket", &object, &bytes).await;
+    }
+    let config = ScannerCycleBudgetConfig {
+        max_objects: Some(8),
+        max_entries: Some(8),
+        ..Default::default()
+    };
+    // 41 root and 40 prefix entries at 8 per round, then 80 objects at the 7
+    // that survive each object-budget stop. Rebuilding a dropped index costs
+    // a full prefix enumeration per classification round and overruns this.
+    const ROUND_BOUND: usize = 6 + 5 + 12 + 2;
+    let finished = run_until_finished(&disk_root, cache, config, ROUND_BOUND).await;
+    let totals = finished.checked_flatten("bucket").expect("bucket root");
+    let expected = u64::try_from(WIDTH * 2).expect("fixture size");
+    assert_eq!((totals.objects as u64, totals.versions as u64), (expected, expected));
+    assert!(finished.info.scan_raw_enumeration_page_index.is_none());
+    assert!(finished.info.scan_raw_enumeration_path_page_indexes.is_empty());
+}
+
+/// A name-byte share stops a listing the same way an entry share does, and the
+/// bytes kept for the next restart grow only by what each round charged.
+#[tokio::test]
+#[serial]
+async fn fair_prefix_quantum_name_bytes_bound_each_round_and_its_checkpoint_growth() {
+    const OBJECTS: usize = 64;
+    const NAME_BYTES_PER_ROUND: u64 = 88;
+    const INDEX_BYTES_PER_NAME_BYTE: usize = 16;
+    let temp = tempfile::tempdir().expect("fixture directory");
+    let disk_root = temp.path().join("disk");
+    let mut cache = seed_flat_bucket(&disk_root, OBJECTS).await;
+    let config = ScannerCycleBudgetConfig {
+        max_entry_name_bytes: Some(NAME_BYTES_PER_ROUND),
+        ..Default::default()
+    };
+    let mut indexed = 0;
+    let mut encoded_before = cache.marshal_msg().expect("encode initial cache").len();
+    for _ in 0..8 {
+        let result = scan_one_restart_round(&disk_root, cache, config).await;
+        assert!(!result.finished);
+        assert_eq!(result.observed_entries, 8, "each entry name in this fixture is 11 bytes");
+        cache = result.outcome;
+        let committed = cache
+            .validated_raw_enumeration_page_index()
+            .expect("durable index")
+            .committed_entries()
+            .expect("committed entries")
+            .len();
+        assert_eq!(committed, indexed + 8, "every charged entry must become durable");
+        indexed = committed;
+        let encoded = cache.marshal_msg().expect("encode cache").len();
+        let growth = encoded.saturating_sub(encoded_before);
+        assert!(
+            growth <= usize::try_from(NAME_BYTES_PER_ROUND).expect("fixture size") * INDEX_BYTES_PER_NAME_BYTE,
+            "checkpoint grew {growth} bytes in one round"
+        );
+        encoded_before = encoded;
+    }
+}
+
+/// Stopping for an entry share must leave the same exact totals a full sweep
+/// produces, and a Deep request must not inherit Normal raw coverage.
+#[tokio::test]
+#[serial]
+async fn fair_prefix_quantum_budget_stop_keeps_progress_without_certifying_it() {
+    const OBJECTS: usize = 24;
+    let temp = tempfile::tempdir().expect("fixture directory");
+    let disk_root = temp.path().join("disk");
+    let cache = seed_flat_bucket(&disk_root, OBJECTS).await;
+    let config = ScannerCycleBudgetConfig {
+        max_objects: Some(4),
+        max_entries: Some(4),
+        ..Default::default()
+    };
+    let round = scan_one_restart_round(&disk_root, cache, config).await;
+    assert!(!round.finished);
+    assert!(!round.outcome.info.snapshot_complete, "budget stops are progress-only");
+    assert!(round.outcome.info.scan_plan_digest.is_none(), "a partial sweep must not be reusable");
+    assert!(round.outcome.has_scan_position(), "the saved cache must carry the resume position");
+
+    let mut deep = round.outcome.clone();
+    let mut identity = deep.info.scan_identity.expect("checkpoint identity");
+    identity.scan_mode = HealScanMode::Deep;
+    let plan = crate::data_usage_define::DataUsageScanPlanDigest([31; 32]);
+    let source = crate::data_usage_define::DataUsageCacheSource::new(0, 0);
+    deep.prepare_bucket_checkpoint("bucket", 2, 0, source, plan, identity);
+    assert!(!deep.has_scan_position(), "Normal raw coverage must not carry into a Deep sweep");
+    assert!(deep.info.scan_raw_enumeration_path_page_indexes.is_empty());
+
+    let finished = run_until_finished(&disk_root, round.outcome, config, 16).await;
+    let totals = finished.checked_flatten("bucket").expect("bucket root");
+    let expected = u64::try_from(OBJECTS).expect("fixture size");
+    assert_eq!((totals.objects as u64, totals.versions as u64), (expected, expected));
+}

crates/scanner/src/scanner_folder/tests/node_admission.rs

@@ -497,6 +497,7 @@ async fn scanner_node_admission_mid_walk_shrink_preserves_partial_and_resumes_al
         panic!("deferral below root cannot certify completion: {partial:?}");
     };
     assert!(!partial.info.snapshot_complete);
+    assert!(partial.info.scan_resource_deferred);
     assert_eq!(
         partial
             .size_recursive("bucket")
@@ -610,6 +611,7 @@ async fn check_early_admission_deferral(node_open: bool, physical_limit: u32) {
         panic!("resource pressure is a partial census, not a disk error: {result:?}");
     };
     assert!(!partial.info.snapshot_complete);
+    assert!(partial.info.scan_resource_deferred);
     let totals = partial.size_recursive("bucket").expect("old tree retained");
     assert_eq!((totals.objects, totals.size), (7, 19));
     assert_eq!(gate.completed(), Some(0));

crates/scanner/src/scanner_io.rs

@@ -18,9 +18,9 @@ use crate::scanner_budget::ScannerCycleBudget;
 use crate::scanner_folder::{ScannerBucketPrefixScanScope, ScannerItem, scan_data_folder_scoped};
 use crate::sleeper::SCANNER_SLEEPER;
 use crate::{
-    DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, DataUsageCache, DataUsageCacheInfo, DataUsageCachePrepareOutcome,
-    DataUsageCacheSource, DataUsageEntry, DataUsageEntryInfo, DataUsageInfo, DataUsageScanPlanDigest, DataUsageSnapshotSetState,
-    ScannerError, SizeSummary, TierStats,
+    DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, DataUsageBucketServiceLedger, DataUsageCache, DataUsageCacheInfo,
+    DataUsageCachePrepareOutcome, DataUsageCacheSource, DataUsageEntry, DataUsageEntryInfo, DataUsageInfo,
+    DataUsageScanPlanDigest, DataUsageSnapshotSetState, ScannerError, SizeSummary, TierStats,
 };
 use bytes::Bytes;
 use futures::future::join_all;

crates/scanner/src/scanner_io/cache.rs

@@ -1200,6 +1200,84 @@ fn log_missing_set_snapshot_metadata(store: &SetDisks, info: &DataUsageCacheInfo
     );
 }
 
+/// Saves a set's service order over its unchanged cache, which an incomplete
+/// cycle otherwise never writes. Under the set cache lock and its revisions the
+/// save cannot replace newer coverage; a failed save costs fairness only.
+pub(super) async fn persist_scanner_service_ledger(
+    store: Arc<SetDisks>,
+    source: DataUsageCacheSource,
+    mut ledger: DataUsageBucketServiceLedger,
+    cycle: u64,
+    leader_epoch: u64,
+    expected_publication_epoch: u64,
+) -> bool {
+    let log_skipped = |state: &str, error: &dyn std::fmt::Display| {
+        warn!(
+            target: "rustfs::scanner::io",
+            event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+            component = LOG_COMPONENT_SCANNER,
+            subsystem = LOG_SUBSYSTEM_IO,
+            state,
+            pool = source.pool_index,
+            set = source.set_index,
+            cycle,
+            leader_epoch,
+            cache_name = DATA_USAGE_CACHE_NAME,
+            error = %error,
+            "Scanner kept the previous bucket service order"
+        );
+    };
+    let guard = match acquire_scanner_cache_locks(store.as_ref(), DATA_USAGE_CACHE_NAME, source).await {
+        Ok(guard) => guard,
+        Err(error) => {
+            log_skipped(error.state(), &error);
+            return false;
+        }
+    };
+    let mut persisted = DataUsageCache::default();
+    let revisions = match persisted.load_with_revisions(store.clone(), DATA_USAGE_CACHE_NAME).await {
+        Ok(revisions) => revisions,
+        Err(error) => {
+            log_skipped("service_order_load_failed", &error);
+            return false;
+        }
+    };
+    if persisted.info.next_cycle > cycle
+        || persisted.info.leader_epoch > leader_epoch
+        || persisted
+            .info
+            .scan_bucket_service
+            .as_ref()
+            .is_some_and(|saved| saved.leader_epoch > leader_epoch)
+    {
+        log_skipped("service_order_stale", &"a newer scanner owns the set cache");
+        return false;
+    }
+    if guard.is_lock_lost() {
+        log_skipped("service_order_lock_lost", &"set cache lock was lost");
+        return false;
+    }
+    if persisted.info.name.is_empty() {
+        // First incomplete cycle of a set: the order is its only content.
+        persisted.info.name = DATA_USAGE_ROOT.to_string();
+        persisted.info.source = Some(source);
+        persisted.info.leader_epoch = leader_epoch;
+        persisted.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT;
+    }
+    ledger.leader_epoch = leader_epoch;
+    persisted.info.scan_bucket_service = Some(ledger);
+    match persisted
+        .save_with_revisions_for_epoch(store, DATA_USAGE_CACHE_NAME, &revisions, expected_publication_epoch)
+        .await
+    {
+        Ok(()) => true,
+        Err(error) => {
+            log_skipped("service_order_save_failed", &error);
+            false
+        }
+    }
+}
+
 pub(super) async fn persist_and_publish_cache_snapshot(
     store: Arc<SetDisks>,
     updates: &mpsc::Sender<DataUsageCache>,

crates/scanner/src/scanner_io/dirty_usage.rs

@@ -1638,6 +1638,7 @@ pub(super) async fn requeue_bucket_work(
     bucket: &BucketInfo,
     work_guard: &mut BucketWorkGuard,
 ) -> bool {
+    work_guard.settle_service_for_requeue();
     if bucket_tx.send(bucket.clone()).await.is_err() {
         return false;
     }

crates/scanner/src/scanner_io/guards.rs

@@ -307,6 +307,33 @@ impl ScannerServiceCohort {
     }
 }
 
+/// Publishes one aggregate per set: each failure is already logged where it
+/// happened, and this keeps the unpaid total visible after a restart.
+pub(super) fn record_bucket_failure_debt(source: DataUsageCacheSource, cycle: u64, ledger: &DataUsageBucketServiceLedger) {
+    let (debtors, first_bucket) = ledger.failure_debt();
+    metrics::gauge!(
+        "rustfs_scanner_bucket_failure_debt",
+        "pool" => source.pool_index.to_string(),
+        "set" => source.set_index.to_string()
+    )
+    .set(debtors as f64);
+    if debtors > 0 {
+        warn!(
+            target: "rustfs::scanner::io",
+            event = EVENT_SCANNER_SET_STATE,
+            component = LOG_COMPONENT_SCANNER,
+            subsystem = LOG_SUBSYSTEM_IO,
+            state = "bucket_failure_debt",
+            pool = source.pool_index,
+            set = source.set_index,
+            cycle,
+            debtors,
+            first_bucket,
+            "Scanner buckets still owe a successful sweep"
+        );
+    }
+}
+
 pub(super) async fn wait_for_bucket_scan_permit(
     semaphore: &Arc<Semaphore>,
     ctx: &CancellationToken,
@@ -408,10 +435,102 @@ pub(super) struct DiskBucketScanActiveGuard {
     set: String,
 }
 
+#[derive(Default)]
+struct ScannerBucketServiceRoundState {
+    ledger: DataUsageBucketServiceLedger,
+    claimed: HashMap<String, u64>,
+    interrupted: HashSet<String>,
+}
+
+/// One cycle's view of a set's durable service order. Lock order: taken alone,
+/// and never held across an await.
+pub(super) struct ScannerBucketServiceRound(StdMutex<ScannerBucketServiceRoundState>);
+
+impl ScannerBucketServiceRound {
+    pub(super) fn new(ledger: DataUsageBucketServiceLedger) -> Self {
+        Self(StdMutex::new(ScannerBucketServiceRoundState {
+            ledger,
+            ..Default::default()
+        }))
+    }
+
+    fn state(&self) -> std::sync::MutexGuard<'_, ScannerBucketServiceRoundState> {
+        self.0.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    pub(super) fn order_buckets(&self, buckets: &mut [BucketInfo]) {
+        // The stable sort keeps the caller's order for equal positions.
+        let state = self.state();
+        buckets.sort_by_key(|bucket| state.ledger.wait_position(&bucket.name));
+    }
+
+    fn claim(&self, bucket: &str) {
+        let mut state = self.state();
+        if let Some(previous) = state.ledger.record_service(bucket) {
+            // A second claim in one cycle keeps the position before the first.
+            state.claimed.entry(bucket.to_owned()).or_insert(previous);
+        }
+    }
+
+    fn release(&self, bucket: &str) {
+        let mut state = self.state();
+        if let Some(previous) = state.claimed.remove(bucket) {
+            state.ledger.return_service(bucket, previous);
+        }
+    }
+
+    fn interrupt(&self, bucket: &str) {
+        self.state().interrupted.insert(bucket.to_owned());
+    }
+
+    /// Closes the cycle. Only a real worker completion clears debt; a claimed
+    /// bucket that failed or stayed partial adds debt unless a budget stopped
+    /// it. `None` means no bucket was served, so nothing needs persisting.
+    pub(super) fn settle(
+        &self,
+        completed: &HashSet<String>,
+        failed: &HashSet<String>,
+        partial: &HashSet<String>,
+    ) -> Option<DataUsageBucketServiceLedger> {
+        let mut state = self.state();
+        let state = &mut *state;
+        if state.claimed.is_empty() {
+            return None;
+        }
+        for bucket in state.claimed.keys() {
+            if completed.contains(bucket) {
+                state.ledger.record_certified(bucket);
+            } else if (failed.contains(bucket) || partial.contains(bucket)) && !state.interrupted.contains(bucket) {
+                state.ledger.record_failure(bucket);
+            }
+        }
+        state.claimed.clear();
+        Some(std::mem::take(&mut state.ledger))
+    }
+}
+
+struct BucketServiceClaim {
+    round: Arc<ScannerBucketServiceRound>,
+    bucket: String,
+    ctx: CancellationToken,
+    budget: Arc<ScannerCycleBudget>,
+}
+
+impl BucketServiceClaim {
+    fn settle(self, requeued: bool) {
+        if self.budget.budget_elapsed() || self.ctx.is_cancelled() {
+            self.round.interrupt(&self.bucket);
+        } else if requeued {
+            self.round.release(&self.bucket);
+        }
+    }
+}
+
 pub(super) struct BucketWorkGuard {
     remaining: Arc<AtomicUsize>,
     complete: CancellationToken,
     requeued: bool,
+    service: Option<BucketServiceClaim>,
 }
 
 impl BucketWorkGuard {
@@ -420,16 +539,53 @@ impl BucketWorkGuard {
             remaining,
             complete,
             requeued: false,
+            service: None,
         }
     }
 
     pub(super) fn mark_requeued(&mut self) {
         self.requeued = true;
     }
+
+    /// Settles the claim before the bucket is visible to another worker, whose
+    /// own claim a later release would otherwise undo.
+    pub(super) fn settle_service_for_requeue(&mut self) {
+        if let Some(claim) = self.service.take() {
+            claim.settle(true);
+        }
+    }
+
+    /// Records a stop that is neither service nor failure, such as a resource veto.
+    pub(super) fn mark_service_interrupted(&self) {
+        if let Some(claim) = &self.service {
+            claim.round.interrupt(&claim.bucket);
+        }
+    }
+
+    /// Spends this bucket's service turn. The guard settles it at the end of
+    /// the admission: a requeue without a spent budget hands the turn back.
+    pub(super) fn claim_service(
+        &mut self,
+        round: Arc<ScannerBucketServiceRound>,
+        bucket: &str,
+        ctx: CancellationToken,
+        budget: Arc<ScannerCycleBudget>,
+    ) {
+        round.claim(bucket);
+        self.service = Some(BucketServiceClaim {
+            round,
+            bucket: bucket.to_owned(),
+            ctx,
+            budget,
+        });
+    }
 }
 
 impl Drop for BucketWorkGuard {
     fn drop(&mut self) {
+        if let Some(claim) = self.service.take() {
+            claim.settle(self.requeued);
+        }
         if !self.requeued && self.remaining.fetch_sub(1, Ordering::AcqRel) == 1 {
             self.complete.cancel();
         }

crates/scanner/src/scanner_io/io_cache.rs

@@ -348,10 +348,21 @@ impl ScannerIOCache for SetDisks {
                 None
             }
         };
+        let persisted_bucket_service = old_cache
+            .info
+            .scan_bucket_service
+            .take()
+            .filter(|_| old_cache.info.source == Some(source));
         // Captured before `prepare_for_scan` clears the mark: only a complete
         // snapshot may stand in for a bucket whose walk stops early.
         let old_cache_certified = old_cache.info.snapshot_complete && old_cache.info.last_update.is_some();
         let current_bucket_incarnations = scanner_current_bucket_incarnations(self.as_ref(), &all_buckets).await;
+        let service_round = Arc::new(ScannerBucketServiceRound::new(DataUsageBucketServiceLedger::restore(
+            clone_cache_value(&persisted_bucket_service),
+            expected_publication_epoch,
+            all_buckets.iter().map(|bucket| bucket.name.as_str()),
+            current_bucket_incarnations.as_ref(),
+        )));
         let scoped_scan = prepare_scoped_set_scan(
             &old_cache,
             &buckets,
@@ -412,6 +423,7 @@ impl ScannerIOCache for SetDisks {
             };
             cache.info.last_update = Some(now);
             cache.info.snapshot_complete = true;
+            cache.info.scan_bucket_service = persisted_bucket_service;
             let replication_execution = super::cache::scanner_replication_execution_digest(execution_digest, &cache);
             cache.info.scan_execution_digest = Some(replication_execution);
             cache.info.segment_invalidation_proof = completed_segment_invalidation_proof;
@@ -753,6 +765,9 @@ impl ScannerIOCache for SetDisks {
                 .unwrap_or_else(|poisoned| poisoned.into_inner())
                 .order_buckets(source, &mut scan_order);
         }
+        // Durable age outranks both orders above; they only break its ties,
+        // so a restart cannot hand the budget back to the same first buckets.
+        service_round.order_buckets(&mut scan_order);
 
         for bucket in scan_order.iter() {
             if let Some(c) = old_cache.find(&bucket.name) {
@@ -790,6 +805,7 @@ impl ScannerIOCache for SetDisks {
         } in walkers
         {
             let service_cohort_clone = service_cohort.clone();
+            let service_round_clone = service_round.clone();
             let bucket_rx_mutex_clone = bucket_rx_mutex.clone();
             let bucket_tx_clone = bucket_tx.clone();
             let remaining_bucket_work_clone = remaining_bucket_work.clone();
@@ -874,6 +890,13 @@ impl ScannerIOCache for SetDisks {
                             .unwrap_or_else(|poisoned| poisoned.into_inner())
                             .record_admitted(source, &bucket.name);
                     }
+                    let (bucket_budget, scan_ctx) = budget_clone.for_bucket(&ctx_clone);
+                    work_guard.claim_service(
+                        service_round_clone.clone(),
+                        &bucket.name,
+                        ctx_clone.clone(),
+                        bucket_budget.clone(),
+                    );
 
                     debug!(
                         target: "rustfs::scanner::io",
@@ -915,8 +938,8 @@ impl ScannerIOCache for SetDisks {
                         remote_session_sequence = next_sequence;
                         let remote_scan = crate::remote_scanner::scan_remote_bucket(
                             &disk,
-                            ctx_clone.clone(),
-                            budget_clone.clone(),
+                            scan_ctx.clone(),
+                            bucket_budget.clone(),
                             crate::remote_scanner::RemoteScannerScanSpec {
                                 bucket: &bucket.name,
                                 next_cycle: want_cycle,
@@ -959,6 +982,11 @@ impl ScannerIOCache for SetDisks {
                                 }
                             }
                             Ok(crate::remote_scanner::RemoteScannerOutcome::Partial) => {
+                                // A remote count budget stops on entries it does
+                                // not report back, so the partial proves no failure.
+                                if bucket_budget.max_objects().is_some() || bucket_budget.max_directories().is_some() {
+                                    work_guard.mark_service_interrupted();
+                                }
                                 record_partial_dirty_bucket(&partial_dirty_buckets_clone, &bucket.name).await;
                                 debug!(
                                     target: "rustfs::scanner::io",
@@ -1096,6 +1124,11 @@ impl ScannerIOCache for SetDisks {
                     };
                     #[cfg(test)]
                     super::tests::observe_coordinator_bucket_scan_admitted(worker_index, &bucket.name).await;
+                    #[cfg(test)]
+                    if super::tests::coordinator_bucket_scan_fails(&bucket.name) {
+                        record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
+                        continue;
+                    }
 
                     // Lock order: scanner leader fence -> set-scoped per-bucket cache lock ->
                     // cache object read/write.
@@ -1331,7 +1364,7 @@ impl ScannerIOCache for SetDisks {
                     let (checkpoint_tx, mut checkpoint_rx) = mpsc::channel::<DataUsageCache>(1);
                     let (node_tx, mut node_rx) = crate::workload_admission::scanner_node_request_channel();
 
-                    let scan_ctx = ctx_clone.child_token();
+                    let scan_position_before = cache.durable_scan_position();
                     #[cfg(feature = "e2e-test-hooks")]
                     {
                         let disk_uuid = if crate::e2e_barrier::worker_observation::enabled() {
@@ -1351,7 +1384,7 @@ impl ScannerIOCache for SetDisks {
                     }
                     let scan = crate::workload_admission::with_scanner_node_requests(node_tx, disk.clone().nsscanner_disk(
                         scan_ctx.clone(),
-                        budget_clone.clone(),
+                        bucket_budget.clone(),
                         set_disk_inventory_clone.as_ref().clone(),
                         cache,
                         None,
@@ -1501,6 +1534,20 @@ impl ScannerIOCache for SetDisks {
                     }
                     let scan_outcome = match scan_result {
                         Ok(scan_outcome) => scan_outcome,
+                        Err(e) if bucket_budget.quantum_exhausted() => {
+                            record_partial_dirty_bucket(&partial_dirty_buckets_clone, &bucket.name).await;
+                            debug!(
+                                target: "rustfs::scanner::io",
+                                event = EVENT_SCANNER_DISK_BUCKET_STATE,
+                                component = LOG_COMPONENT_SCANNER,
+                                subsystem = LOG_SUBSYSTEM_IO,
+                                bucket = %bucket.name,
+                                state = "quantum_exhausted_without_progress",
+                                error = %e,
+                                "Scanner bucket quantum ended before durable progress"
+                            );
+                            continue;
+                        }
                         Err(e) => {
                             record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
                             if ctx_clone.is_cancelled() {
@@ -1542,7 +1589,12 @@ impl ScannerIOCache for SetDisks {
                             }
                             (completed_cache, false)
                         }
-                        ScannerDiskScanOutcome::Partial(partial_cache) => (partial_cache, true),
+                        ScannerDiskScanOutcome::Partial(partial_cache) => {
+                            if partial_cache.info.scan_resource_deferred {
+                                work_guard.mark_service_interrupted();
+                            }
+                            (partial_cache, true)
+                        }
                         ScannerDiskScanOutcome::NamespaceNotFound(_) => {
                             if requeue_bucket_work(&bucket_tx_clone, &bucket, &mut work_guard).await {
                                 increment_disk_bucket_scans_queued(
@@ -1566,7 +1618,6 @@ impl ScannerIOCache for SetDisks {
                         }
                     };
                     if partial {
-                        record_partial_dirty_bucket(&partial_dirty_buckets_clone, &bucket.name).await;
                         let done_save = Metrics::time(Metric::SaveUsage);
                         let partial_saved = if cache_guard.is_lock_lost() {
                             false
@@ -1604,6 +1655,21 @@ impl ScannerIOCache for SetDisks {
                         if partial_saved {
                             log_saved_partial_bucket_cache(&cache, &cache_name);
                         }
+                        // Only saved forward motion earns another turn in this
+                        // cycle; otherwise the bucket waits for the next one.
+                        if partial_saved
+                            && bucket_budget.quantum_exhausted()
+                            && cache.durable_scan_position() != scan_position_before
+                            && requeue_bucket_work(&bucket_tx_clone, &bucket, &mut work_guard).await
+                        {
+                            increment_disk_bucket_scans_queued(
+                                &queued_disk_bucket_scans_clone,
+                                &pool_label_clone,
+                                &set_label_clone,
+                            );
+                            continue;
+                        }
+                        record_partial_dirty_bucket(&partial_dirty_buckets_clone, &bucket.name).await;
 
                         continue;
                     }
@@ -1824,6 +1890,14 @@ impl ScannerIOCache for SetDisks {
                 .await
                 .extend(carried_buckets.iter().cloned());
         }
+        let service_ledger = {
+            let failed = bucket_failures.hard.lock().await;
+            let partial = bucket_failures.partial.lock().await;
+            service_round.settle(&completed_buckets, &failed, &partial)
+        };
+        if let Some(ledger) = &service_ledger {
+            record_bucket_failure_debt(source, want_cycle, ledger);
+        }
         let completed_count = completed_buckets.len();
         let covered_count = completed_count.saturating_add(carried_buckets.len());
         let coverage_complete =
@@ -1851,6 +1925,7 @@ impl ScannerIOCache for SetDisks {
             if cold_zero_walk_reuse_candidate {
                 cold_zero_walk_reuse_observed.store(true, Ordering::Release);
             }
+            cache.info.scan_bucket_service = service_ledger.or(persisted_bucket_service);
             let _ = persist_and_publish_cache_snapshot(
                 self.clone(),
                 &updates,
@@ -1908,6 +1983,20 @@ impl ScannerIOCache for SetDisks {
                 cancelled = ctx.is_cancelled(),
                 "Scanner set cache publish skipped because cycle did not complete cleanly"
             );
+            // Publish the scope first: a contended cache lock must not delay it.
+            // A stopped owner leaves the set cache untouched; only a budget stop
+            // or a finished incomplete cycle hands its order to the next one.
+            if let Some(ledger) = service_ledger.filter(|_| !ctx.is_cancelled() || budget.budget_elapsed()) {
+                persist_scanner_service_ledger(
+                    self.clone(),
+                    source,
+                    ledger,
+                    want_cycle,
+                    leader_epoch,
+                    expected_publication_epoch,
+                )
+                .await;
+            }
         }
 
         debug!(

crates/scanner/src/scanner_io/tests.rs

@@ -45,6 +45,7 @@ use time::OffsetDateTime;
 use uuid::Uuid;
 
 mod checkpoint_coverage_identity;
+mod cold_bucket_service;
 mod collector_lifecycle;
 mod cycle_collector_diagnostics;
 mod expiry_lock;
@@ -5176,6 +5177,35 @@ pub(super) async fn observe_coordinator_bucket_scan_admitted(worker_index: usize
     probe.active.fetch_sub(1, Ordering::SeqCst);
 }
 
+static FAILING_COORDINATOR_BUCKETS: std::sync::LazyLock<StdMutex<HashSet<String>>> =
+    std::sync::LazyLock::new(|| StdMutex::new(HashSet::new()));
+
+pub(super) fn coordinator_bucket_scan_fails(bucket: &str) -> bool {
+    FAILING_COORDINATOR_BUCKETS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .contains(bucket)
+}
+
+struct FailingCoordinatorBucket(String);
+
+impl Drop for FailingCoordinatorBucket {
+    fn drop(&mut self) {
+        FAILING_COORDINATOR_BUCKETS
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .remove(&self.0);
+    }
+}
+
+fn fail_coordinator_bucket_scan(bucket: &str) -> FailingCoordinatorBucket {
+    FAILING_COORDINATOR_BUCKETS
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .insert(bucket.to_string());
+    FailingCoordinatorBucket(bucket.to_string())
+}
+
 pub(super) async fn observe_coordinator_bucket_worker_started(worker_index: usize) {
     collector_lifecycle::observe_worker_start(worker_index).await;
 }

crates/scanner/src/scanner_io/tests/checkpoint_coverage_identity.rs

@@ -94,15 +94,15 @@ async fn checkpoint_resumes_across_a_maintenance_publication() {
     assert!(result.clear_verified_usage().is_empty());
     assert!(dirty_usage_buckets_for_tests().is_empty());
 
-    // Cycle 2: the written bucket is the only dirty one, so it is walked
-    // first and the object budget stops its sweep with a checkpoint; the
-    // cycle itself is incomplete.
+    // Cycle 2: the object budget stops the written bucket's sweep with a
+    // checkpoint; the cycle itself is incomplete. Service order follows the
+    // wait, so the budget leaves it two objects even behind the idle bucket.
     for object in ["p6/g", "p7/h"] {
         put_fixture_object(&store, "hot-bucket", object).await;
     }
     record_dirty_usage_bucket("hot-bucket");
     wait_for_namespace_commit_tails(store.as_ref()).await;
-    let (result, _, budget) = run_full_cycle(&store, 2, Some(2)).await;
+    let (result, _, budget) = run_full_cycle(&store, 2, Some(3)).await;
     assert_eq!(result.status, ScannerCycleStatus::Incomplete);
     assert!(budget.budget_elapsed());
     let set = Arc::clone(&store.pools[0].disk_set[0]);

crates/scanner/src/scanner_io/tests/cold_bucket_service.rs

@@ -0,0 +1,380 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use std::path::Path;
+
+const BUCKETS: usize = 17;
+const QUANTA: usize = 4;
+const ROUND_BOUND: usize = BUCKETS.div_ceil(QUANTA);
+const HOT: &str = "bucket-00";
+const FAILING: &str = "bucket-01";
+const ROUND_REQUEST_ENV: &str = "RUSTFS_COLD_BUCKET_SERVICE_ROUND";
+const EPOCH: u64 = 9;
+
+fn inventory() -> Vec<BucketInfo> {
+    (0..BUCKETS)
+        .map(|index| BucketInfo {
+            name: format!("bucket-{index:02}"),
+            ..Default::default()
+        })
+        .collect()
+}
+
+fn restore(persisted: Option<DataUsageBucketServiceLedger>, buckets: &[BucketInfo]) -> DataUsageBucketServiceLedger {
+    DataUsageBucketServiceLedger::restore(persisted, EPOCH, buckets.iter().map(|bucket| bucket.name.as_str()), None)
+}
+
+/// One scheduler lifetime over the production order and queue accounting.
+/// `persisted` is decoded from durable bytes; the result is encoded again.
+fn serve_round(persisted: &[u8]) -> (Vec<u8>, Vec<String>) {
+    let buckets = inventory();
+    let decoded = DataUsageCache::unmarshal(persisted).expect("decode saved order");
+    let round = Arc::new(ScannerBucketServiceRound::new(restore(decoded.info.scan_bucket_service, &buckets)));
+    let dirty = DirtyUsageBuckets::from([(HOT.to_string(), 1)]);
+    let mut order = bucket_usage_scan_order(&buckets, &DataUsageCache::default(), &dirty);
+    round.order_buckets(&mut order);
+    let served = order
+        .iter()
+        .take(QUANTA)
+        .map(|bucket| bucket.name.clone())
+        .collect::<Vec<_>>();
+    let mut completed = HashSet::new();
+    let mut failed = HashSet::new();
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&ctx, Default::default());
+    for bucket in &served {
+        let mut work = BucketWorkGuard::new(Arc::new(AtomicUsize::new(2)), CancellationToken::new());
+        work.claim_service(round.clone(), bucket, ctx.clone(), budget.clone());
+        drop(work);
+        if bucket == FAILING {
+            failed.insert(bucket.clone());
+        } else {
+            completed.insert(bucket.clone());
+        }
+    }
+    let ledger = round.settle(&completed, &failed, &HashSet::new()).expect("served buckets");
+    let mut cache = DataUsageCache::default();
+    cache.info.scan_bucket_service = Some(ledger);
+    (cache.marshal_msg().expect("encode saved order"), served)
+}
+
+fn assert_service_bound(rounds: &[Vec<String>]) {
+    for bucket in inventory() {
+        let mut waited = 0;
+        for (index, served) in rounds.iter().enumerate() {
+            waited = if served.contains(&bucket.name) { 0 } else { waited + 1 };
+            assert!(waited < ROUND_BOUND, "{} waited {waited} rounds at round {index}", bucket.name);
+        }
+    }
+}
+
+#[test]
+fn cold_bucket_service_bound_holds_through_the_saved_codec() {
+    let mut saved = DataUsageCache::default().marshal_msg().expect("empty cache");
+    let mut rounds = Vec::new();
+    for _ in 0..(ROUND_BOUND * 4) {
+        let (next, served) = serve_round(&saved);
+        saved = next;
+        rounds.push(served);
+    }
+    assert_service_bound(&rounds);
+    let ledger = DataUsageCache::unmarshal(&saved)
+        .expect("decode final order")
+        .info
+        .scan_bucket_service
+        .expect("saved order");
+    let failing = ledger.buckets[FAILING];
+    assert!(failing.failure_debt >= 4, "every failed turn must stay visible: {failing:?}");
+    assert_eq!(failing.last_certified, None, "failure is not certification");
+    assert!(failing.service_quanta >= 4, "a failing bucket still spends real turns");
+    assert_eq!(ledger.failure_debt(), (1, Some(FAILING)));
+    assert_eq!(ledger.buckets[HOT].failure_debt, 0);
+    let hot_quanta = ledger.buckets[HOT].service_quanta;
+    assert!(
+        ledger
+            .buckets
+            .values()
+            .all(|service| service.service_quanta.abs_diff(hot_quanta) <= 1),
+        "a hot bucket must not earn more turns than a cold one: {ledger:?}"
+    );
+}
+
+/// The driver and each round are separate OS processes, so the only carrier
+/// between rounds is the saved cache file.
+#[test]
+fn cold_bucket_service_bound_holds_across_fresh_os_processes() {
+    let temp = tempfile::tempdir().expect("fixture directory");
+    let order = temp.path().join("order.bin");
+    std::fs::write(&order, DataUsageCache::default().marshal_msg().expect("empty cache")).expect("seed order");
+    let test = format!(
+        "{}::cold_bucket_service_process_round_worker",
+        module_path!().split_once("::").expect("crate path").1
+    );
+    let binary = std::env::current_exe().expect("test binary");
+    let listed = std::process::Command::new(&binary)
+        .args([test.as_str(), "--exact", "--list"])
+        .output()
+        .expect("list worker");
+    assert!(
+        String::from_utf8_lossy(&listed.stdout).contains(&format!("{test}: test")),
+        "the restart driver must select a real worker"
+    );
+    let mut rounds = Vec::new();
+    let mut pids = HashSet::new();
+    for round in 0..(ROUND_BOUND * 2) {
+        let status = std::process::Command::new(&binary)
+            .args([test.as_str(), "--exact", "--test-threads=1"])
+            .env(ROUND_REQUEST_ENV, &order)
+            .stdout(std::process::Stdio::null())
+            .status()
+            .expect("start worker round");
+        assert!(status.success(), "worker round {round} failed");
+        let report = std::fs::read_to_string(order.with_extension("served")).expect("worker report");
+        let (pid, served) = report.split_once('\n').expect("worker report shape");
+        assert!(pids.insert(pid.to_string()), "round {round} reused a process");
+        rounds.push(served.lines().map(str::to_owned).collect::<Vec<_>>());
+    }
+    assert_service_bound(&rounds);
+}
+
+#[test]
+fn cold_bucket_service_process_round_worker() {
+    let Some(order) = std::env::var_os(ROUND_REQUEST_ENV) else {
+        let empty = DataUsageCache::default().marshal_msg().expect("empty cache");
+        assert_eq!(serve_round(&empty).1.len(), QUANTA);
+        return;
+    };
+    let order = Path::new(&order);
+    let (saved, served) = serve_round(&std::fs::read(order).expect("read saved order"));
+    std::fs::write(order, saved).expect("save order");
+    std::fs::write(order.with_extension("served"), format!("{}\n{}", std::process::id(), served.join("\n")))
+        .expect("write worker report");
+}
+
+#[test]
+fn cold_bucket_service_new_bucket_joins_behind_existing_waiters() {
+    let buckets = inventory();
+    let mut ledger = restore(None, &buckets);
+    ledger.record_service("bucket-00");
+    let with_new = buckets
+        .iter()
+        .cloned()
+        .chain([BucketInfo {
+            name: "a-new-bucket".to_string(),
+            ..Default::default()
+        }])
+        .collect::<Vec<_>>();
+    let ledger = restore(Some(ledger), &with_new);
+    let mut order = with_new;
+    order.sort_by(|left, right| left.name.cmp(&right.name));
+    ScannerBucketServiceRound::new(ledger.clone()).order_buckets(&mut order);
+    assert_eq!(order[0].name, "bucket-01", "a stream of new buckets must not displace waiters");
+    let arrival = ledger.buckets["a-new-bucket"].waiting_since;
+    assert!(arrival > ledger.buckets["bucket-01"].waiting_since);
+    assert!(
+        arrival < ledger.buckets["bucket-00"].waiting_since,
+        "an arrival must not wait behind a bucket that already had its turn"
+    );
+}
+
+#[test]
+fn cold_bucket_service_scope_change_invalidates_the_saved_order() {
+    let buckets = inventory();
+    let mut ledger = restore(None, &buckets);
+    ledger.record_service(FAILING);
+    ledger.record_failure(FAILING);
+
+    let moved = DataUsageBucketServiceLedger::restore(
+        Some(ledger.clone()),
+        EPOCH + 1,
+        buckets.iter().map(|bucket| bucket.name.as_str()),
+        None,
+    );
+    assert_eq!(moved.publication_epoch, EPOCH + 1);
+    assert_eq!(moved.buckets[FAILING].failure_debt, 0, "another movement epoch starts a new order");
+
+    let mut corrupt = ledger.clone();
+    corrupt.next_sequence = 0;
+    let rebuilt = restore(Some(corrupt), &buckets);
+    assert_eq!(rebuilt.buckets[FAILING].service_quanta, 0, "an impossible wait position is not trusted");
+
+    let remaining = restore(Some(ledger.clone()), &buckets[..1]);
+    assert_eq!(remaining.buckets.len(), 1, "a deleted bucket leaves the order");
+}
+
+#[test]
+fn cold_bucket_service_recreated_bucket_loses_its_old_standing() {
+    let buckets = inventory();
+    let first = Uuid::from_u128(1);
+    let incarnations = HashMap::from([(FAILING.to_string(), first)]);
+    let names = || buckets.iter().map(|bucket| bucket.name.as_str());
+    let mut ledger = DataUsageBucketServiceLedger::restore(None, EPOCH, names(), Some(&incarnations));
+    ledger.record_failure(FAILING);
+    let same = DataUsageBucketServiceLedger::restore(Some(ledger.clone()), EPOCH, names(), Some(&incarnations));
+    assert_eq!(same.buckets[FAILING].failure_debt, 1);
+
+    let unknown = DataUsageBucketServiceLedger::restore(Some(ledger.clone()), EPOCH, names(), None);
+    assert_eq!(
+        unknown.buckets[FAILING].incarnation,
+        Some(first),
+        "a transient identity gap must not erase debt"
+    );
+
+    let recreated = HashMap::from([(FAILING.to_string(), Uuid::from_u128(2))]);
+    let replaced = DataUsageBucketServiceLedger::restore(Some(ledger), EPOCH, names(), Some(&recreated));
+    assert_eq!(replaced.buckets[FAILING].failure_debt, 0, "a recreated bucket does not inherit debt");
+    assert_eq!(replaced.buckets[FAILING].incarnation, Some(Uuid::from_u128(2)));
+}
+
+#[test]
+fn cold_bucket_service_budget_stop_and_resource_veto_add_no_debt() {
+    let buckets = inventory();
+    let ledger = restore(None, &buckets);
+    let before = ledger.buckets["bucket-02"].waiting_since;
+    let round = Arc::new(ScannerBucketServiceRound::new(ledger));
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&ctx, Default::default());
+    let work = || BucketWorkGuard::new(Arc::new(AtomicUsize::new(2)), CancellationToken::new());
+
+    let mut vetoed = work();
+    vetoed.claim_service(round.clone(), "bucket-02", ctx.clone(), budget.clone());
+    vetoed.mark_requeued();
+    drop(vetoed);
+
+    let mut stopped = work();
+    stopped.claim_service(round.clone(), "bucket-03", ctx, budget.clone());
+    budget.cancel_for_runtime();
+    drop(stopped);
+
+    let names = |names: &[&str]| names.iter().map(|name| name.to_string()).collect::<HashSet<_>>();
+    let ledger = round
+        .settle(&HashSet::new(), &names(&["bucket-02", "bucket-03"]), &HashSet::new())
+        .expect("one bucket was served");
+    assert_eq!(ledger.buckets["bucket-02"].waiting_since, before, "a veto hands the turn back");
+    assert_eq!(ledger.buckets["bucket-02"].service_quanta, 0);
+    assert_eq!(ledger.buckets["bucket-03"].service_quanta, 1, "the budget was really spent");
+    assert_eq!(ledger.failure_debt(), (0, None), "budget-cancelled work is progress-only");
+}
+
+#[tokio::test]
+async fn cold_bucket_service_requeue_cannot_undo_the_next_workers_claim() {
+    let buckets = inventory();
+    let ledger = restore(None, &buckets);
+    let before = ledger.buckets["bucket-02"].waiting_since;
+    let round = Arc::new(ScannerBucketServiceRound::new(ledger));
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&ctx, Default::default());
+    let work = || BucketWorkGuard::new(Arc::new(AtomicUsize::new(2)), CancellationToken::new());
+    let (bucket_tx, mut bucket_rx) = tokio::sync::mpsc::channel(1);
+
+    let mut first = work();
+    first.claim_service(round.clone(), "bucket-02", ctx.clone(), budget.clone());
+    assert!(requeue_bucket_work(&bucket_tx, &buckets[2], &mut first).await);
+
+    // The next worker takes the bucket while the first guard is still alive.
+    bucket_rx.recv().await.expect("requeued bucket");
+    let mut second = work();
+    second.claim_service(round.clone(), "bucket-02", ctx, budget);
+    drop(first);
+    drop(second);
+
+    let completed = HashSet::from(["bucket-02".to_string()]);
+    let ledger = round
+        .settle(&completed, &HashSet::new(), &HashSet::new())
+        .expect("the second claim served the bucket");
+    let service = &ledger.buckets["bucket-02"];
+    assert!(service.waiting_since > before, "the completed bucket returns to the tail");
+    assert_eq!(service.service_quanta, 1, "only the completed turn counts");
+}
+
+#[test]
+fn cold_bucket_service_partial_resource_veto_adds_no_debt() {
+    let buckets = inventory();
+    let round = Arc::new(ScannerBucketServiceRound::new(restore(None, &buckets)));
+    let ctx = CancellationToken::new();
+    let budget = ScannerCycleBudget::new(&ctx, Default::default());
+    let work = || BucketWorkGuard::new(Arc::new(AtomicUsize::new(2)), CancellationToken::new());
+
+    let mut vetoed = work();
+    vetoed.claim_service(round.clone(), "bucket-02", ctx.clone(), budget.clone());
+    vetoed.mark_service_interrupted();
+    drop(vetoed);
+    let mut failed = work();
+    failed.claim_service(round.clone(), "bucket-03", ctx, budget);
+    drop(failed);
+
+    let partial = HashSet::from(["bucket-02".to_string(), "bucket-03".to_string()]);
+    let ledger = round
+        .settle(&HashSet::new(), &HashSet::new(), &partial)
+        .expect("two buckets were served");
+    assert_eq!(ledger.buckets["bucket-02"].failure_debt, 0, "a resource block is not a failure");
+    assert_eq!(ledger.buckets["bucket-03"].failure_debt, 1, "an uninterrupted partial stays visible");
+}
+
+#[test]
+fn cold_bucket_service_resource_deferral_marker_is_never_persisted() {
+    let mut cache = DataUsageCache::default();
+    cache.info.name = DATA_USAGE_ROOT.to_string();
+    cache.info.scan_resource_deferred = true;
+    let decoded = DataUsageCache::unmarshal(&cache.marshal_msg().expect("encode")).expect("decode");
+    assert!(!decoded.info.scan_resource_deferred);
+}
+
+#[test]
+fn cold_bucket_service_order_is_an_optional_map_field_for_old_readers() {
+    #[derive(serde::Deserialize)]
+    struct LegacyInfo {
+        name: String,
+        next_cycle: u64,
+    }
+    #[derive(serde::Deserialize)]
+    struct LegacyCache {
+        info: LegacyInfo,
+        #[serde(rename = "cache")]
+        _cache: serde::de::IgnoredAny,
+    }
+    let mut cache = DataUsageCache::default();
+    cache.info.name = DATA_USAGE_ROOT.to_string();
+    cache.info.next_cycle = 7;
+    let absent = cache.marshal_msg().expect("encode cache");
+    let field = b"scan_bucket_service";
+    assert!(
+        !absent.windows(field.len()).any(|window| window == field),
+        "an unused order must not change the encoded cache"
+    );
+
+    let buckets = inventory();
+    let mut ledger = restore(None, &buckets);
+    ledger.record_service(FAILING);
+    ledger.record_failure(FAILING);
+    cache.info.scan_bucket_service = Some(ledger.clone());
+    let encoded = cache.marshal_msg().expect("encode order");
+    let legacy: LegacyCache = rmp_serde::from_slice(&encoded).expect("old readers must ignore the order");
+    assert_eq!((legacy.info.name.as_str(), legacy.info.next_cycle), (DATA_USAGE_ROOT, 7));
+    assert_eq!(
+        DataUsageCache::unmarshal(&encoded)
+            .expect("decode order")
+            .info
+            .scan_bucket_service,
+        Some(ledger)
+    );
+    assert!(
+        DataUsageCache::unmarshal(&absent)
+            .expect("decode a cache from an older writer")
+            .info
+            .scan_bucket_service
+            .is_none()
+    );
+}

crates/scanner/src/scanner_io/tests/service_cohort.rs

@@ -337,3 +337,221 @@ async fn service_cohort_cancelled_dispatch_does_not_consume_waiters_or_leak_perm
     assert_eq!(usage.expect("retry aggregate").objects_total_count, 2);
     clear_dirty_usage_buckets_for_tests();
 }
+
+/// N=17, q=4: every bucket reaches a worker within ceil(17/4)=5 rounds while
+/// one bucket stays hot and another fails forever, with leader-local state
+/// discarded after each round as a restart would.
+#[tokio::test]
+#[serial]
+async fn cold_bucket_service_bound_survives_restart_with_hot_and_failing_bucket() {
+    const BUCKETS: usize = 17;
+    const QUANTA: u64 = 4;
+    const ROUND_BOUND: usize = 5;
+    let (_dir, store) = setup_local_scanner_store(1, 4).await;
+    clear_dirty_usage_buckets_for_tests();
+    let names = (0..BUCKETS).map(|index| format!("cold-{index:02}")).collect::<Vec<_>>();
+    for name in &names {
+        create_cohort_bucket(&store, name).await;
+    }
+    let hot = names[0].clone();
+    let _failing = fail_coordinator_bucket_scan(&names[1]);
+    let mut waited = [0_usize; BUCKETS];
+    for round in 1..=(ROUND_BOUND * 3) {
+        record_dirty_usage_bucket(&hot);
+        let cohort = Arc::new(StdMutex::new(ScannerServiceCohort::default()));
+        let ctx = CancellationToken::new();
+        let budget = ScannerCycleBudget::new_with_progress_tracking(
+            &ctx,
+            ScannerCycleBudgetConfig {
+                max_objects: Some(QUANTA),
+                ..Default::default()
+            },
+        );
+        run_cohort_cycle(&store, cohort.clone(), u64::try_from(round).expect("round"), budget).await;
+        let admitted = cohort
+            .lock()
+            .expect("cohort lock")
+            .admitted_members()
+            .into_iter()
+            .map(|(_, bucket)| bucket)
+            .collect::<HashSet<_>>();
+        for (index, name) in names.iter().enumerate() {
+            waited[index] = if admitted.contains(name) { 0 } else { waited[index] + 1 };
+            assert!(
+                waited[index] < ROUND_BOUND,
+                "{name} waited {} rounds at round {round}; admitted {admitted:?}",
+                waited[index]
+            );
+        }
+    }
+    let saved = load_set_cache(&store).await;
+    let ledger = saved
+        .info
+        .scan_bucket_service
+        .expect("incomplete cycles must save the service order");
+    assert!(ledger.buckets[&names[1]].failure_debt > 0, "the failing bucket must keep visible debt");
+    assert_eq!(ledger.buckets[&names[1]].last_certified, None);
+    assert_eq!(ledger.failure_debt().0, 1, "only the failing bucket owes a sweep");
+    assert!(!saved.info.snapshot_complete, "the saved order must not certify the set");
+    clear_dirty_usage_buckets_for_tests();
+}
+
+async fn load_set_cache(store: &Arc<ECStore>) -> DataUsageCache {
+    let mut cache = DataUsageCache::default();
+    cache
+        .load(store.all_set_disks()[0].clone(), DATA_USAGE_CACHE_NAME)
+        .await
+        .expect("load set cache");
+    cache
+}
+
+fn quantum_budget(quantum: crate::scanner_budget::ScannerBucketQuantum) -> Arc<ScannerCycleBudget> {
+    ScannerCycleBudget::new_with_progress_tracking(
+        &CancellationToken::new(),
+        ScannerCycleBudgetConfig {
+            bucket_quantum: Some(quantum),
+            ..Default::default()
+        },
+    )
+}
+
+#[tokio::test]
+#[serial]
+async fn fair_prefix_quantum_spent_bucket_returns_behind_waiting_buckets() {
+    let (_dir, store) = setup_local_scanner_store(1, 4).await;
+    clear_dirty_usage_buckets_for_tests();
+    create_cohort_bucket_objects(&store, "a-flat", 6).await;
+    create_cohort_bucket(&store, "z-cold-one").await;
+    create_cohort_bucket(&store, "z-cold-two").await;
+    let cohort = Arc::new(StdMutex::new(ScannerServiceCohort::default()));
+    let budget = quantum_budget(crate::scanner_budget::ScannerBucketQuantum {
+        max_objects: Some(2),
+        ..Default::default()
+    });
+    let (result, usage) = run_cohort_cycle(&store, cohort, 1, budget.clone()).await;
+    assert_eq!(result.status, ScannerCycleStatus::Complete, "shares must add up to one whole sweep");
+    assert_eq!(usage.expect("complete aggregate").objects_total_count, 8);
+    assert!(!budget.budget_elapsed(), "only the bucket shares ended");
+
+    let ledger = load_set_cache(&store).await.info.scan_bucket_service.expect("saved order");
+    assert!(ledger.buckets["a-flat"].service_quanta >= 3, "{ledger:?}");
+    assert_eq!(ledger.buckets["z-cold-one"].service_quanta, 1, "{ledger:?}");
+    assert_eq!(ledger.buckets["z-cold-two"].service_quanta, 1, "{ledger:?}");
+    assert!(
+        ledger.buckets["a-flat"].waiting_since > ledger.buckets["z-cold-two"].waiting_since,
+        "the wide bucket must finish behind the buckets it yielded to: {ledger:?}"
+    );
+    assert_eq!(ledger.failure_debt(), (0, None), "a spent share is not a failure");
+    clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+#[serial]
+async fn fair_prefix_quantum_without_durable_motion_ends_the_bucket_turn() {
+    let (_dir, store) = setup_local_scanner_store(1, 4).await;
+    clear_dirty_usage_buckets_for_tests();
+    create_cohort_bucket_objects(&store, "a-flat", 6).await;
+    let cohort = Arc::new(StdMutex::new(ScannerServiceCohort::default()));
+    let budget = quantum_budget(crate::scanner_budget::ScannerBucketQuantum {
+        max_duration: Some(Duration::ZERO),
+        ..Default::default()
+    });
+    let (result, usage) = run_cohort_cycle(&store, cohort, 1, budget).await;
+    assert_eq!(result.status, ScannerCycleStatus::Incomplete);
+    assert!(usage.is_none());
+    let ledger = load_set_cache(&store).await.info.scan_bucket_service.expect("saved order");
+    assert_eq!(
+        ledger.buckets["a-flat"].service_quanta, 1,
+        "a share that saved nothing must not requeue itself: {ledger:?}"
+    );
+    assert_eq!(ledger.failure_debt(), (0, None));
+    clear_dirty_usage_buckets_for_tests();
+}
+
+/// A one-object share stops inside its first child, so the frontier never
+/// moves. Requeuing that bucket would spin until the cycle deadline.
+#[tokio::test]
+#[serial]
+async fn fair_prefix_quantum_repeated_share_without_a_new_frontier_stops_requeuing() {
+    let (_dir, store) = setup_local_scanner_store(1, 4).await;
+    clear_dirty_usage_buckets_for_tests();
+    create_cohort_bucket_objects(&store, "a-flat", 3).await;
+    let cohort = Arc::new(StdMutex::new(ScannerServiceCohort::default()));
+    let budget = quantum_budget(crate::scanner_budget::ScannerBucketQuantum {
+        max_objects: Some(1),
+        ..Default::default()
+    });
+    let (result, usage) = run_cohort_cycle(&store, cohort, 1, budget).await;
+    assert_eq!(result.status, ScannerCycleStatus::Incomplete);
+    assert!(usage.is_none());
+    let ledger = load_set_cache(&store).await.info.scan_bucket_service.expect("saved order");
+    assert!(
+        ledger.buckets["a-flat"].service_quanta <= 3,
+        "turns without durable motion must end: {ledger:?}"
+    );
+    assert_eq!(ledger.failure_debt(), (0, None));
+    clear_dirty_usage_buckets_for_tests();
+}
+
+#[tokio::test]
+#[serial]
+async fn cold_bucket_service_saved_order_keeps_coverage_and_rejects_older_owners() {
+    let (_dir, store) = setup_local_scanner_store(1, 4).await;
+    clear_dirty_usage_buckets_for_tests();
+    create_cohort_bucket(&store, "cold-only").await;
+    let ctx = CancellationToken::new();
+    let cohort = Arc::new(StdMutex::new(ScannerServiceCohort::default()));
+    let (result, _) = run_cohort_cycle(&store, cohort, 3, ScannerCycleBudget::new(&ctx, Default::default())).await;
+    assert_eq!(result.status, ScannerCycleStatus::Complete);
+    let certified = load_set_cache(&store).await;
+    assert!(certified.info.snapshot_complete);
+    let set = store.all_set_disks()[0].clone();
+    let epoch = scanner_publication_epoch(set.clone()).await.expect("publication epoch");
+    let source = DataUsageCacheSource::new(set.pool_index, set.set_index);
+    let mut ledger = certified.info.scan_bucket_service.clone().expect("saved order");
+    ledger.next_sequence += 1;
+    ledger.leader_epoch = 11;
+
+    assert!(persist_scanner_service_ledger(set.clone(), source, ledger.clone(), 3, 11, epoch).await);
+    let saved = load_set_cache(&store).await;
+    assert_eq!(saved.info.scan_bucket_service.as_ref(), Some(&ledger));
+    assert_eq!(saved.info.snapshot_complete, certified.info.snapshot_complete);
+    assert_eq!(saved.info.scan_plan_digest, certified.info.scan_plan_digest);
+    assert_eq!(saved.info.scan_execution_digest, certified.info.scan_execution_digest);
+    assert_eq!(saved.info.next_cycle, certified.info.next_cycle);
+    assert_eq!(saved.info.leader_epoch, certified.info.leader_epoch);
+    assert_eq!(saved.cache.len(), certified.cache.len());
+
+    let mut stale = ledger.clone();
+    stale.next_sequence += 1;
+    assert!(
+        !persist_scanner_service_ledger(set.clone(), source, stale.clone(), 3, 10, epoch).await,
+        "a late write from an older leader must not replace the order"
+    );
+    assert!(
+        !persist_scanner_service_ledger(set.clone(), source, stale, 2, 11, epoch).await,
+        "a late write from an older cycle must not replace the order"
+    );
+    assert_eq!(load_set_cache(&store).await.info.scan_bucket_service.as_ref(), Some(&ledger));
+
+    // An incomplete save leaves the cache stamps at epoch 11, so only the order
+    // itself can reject the leader it replaced.
+    let mut newer = ledger.clone();
+    newer.next_sequence += 1;
+    assert!(persist_scanner_service_ledger(set.clone(), source, newer.clone(), 3, 12, epoch).await);
+    let saved = load_set_cache(&store).await;
+    assert_eq!(saved.info.leader_epoch, certified.info.leader_epoch);
+    assert!(
+        !persist_scanner_service_ledger(set.clone(), source, ledger.clone(), 3, 11, epoch).await,
+        "a late write must not replace an order saved by a newer leader"
+    );
+    assert_eq!(
+        load_set_cache(&store)
+            .await
+            .info
+            .scan_bucket_service
+            .map(|saved| saved.next_sequence),
+        Some(newer.next_sequence)
+    );
+    clear_dirty_usage_buckets_for_tests();
+}

docs/testing/scanner-checkpoint-fixture.md

@@ -16,20 +16,32 @@ Use the `executable` from the scanner library test `compiler-artifact` JSON reco
 python3 scripts/diagnose_scanner_enumeration_restart.py \
   --test-binary /path/to/compiled/scanner-libtest \
   --output /tmp/scanner-enumeration-new-run \
-  --objects 128 --raw-entry-budget 8 --rounds 8
+  --objects 128 --raw-entry-budget 8 --rounds 40
 ```
 
+The round count must cover both stages: at `E` first-seen entries and `O` objects with budget `B`, enumeration takes `E/B` rounds and classification about `O/(B-1)` more, so 128 objects at budget 8 converge in 35 rounds. `--rounds 8` cannot converge at that budget on any correct scanner and reports exit 2 because no round reached classification.
+
 The output directory must not exist. Each round starts a new OS test-worker process, opens the same synthetic disk, decodes the preceding cache, invokes the real scanner, encodes the returned cache, and decodes it again. When cancellation returns no useful partial cache, it preserves the previous cache. Reports identify the actual child PID, round, raw entries and name bytes observed, raw page-index parent/entries, processed/classified objects, retained object/version/byte counts, and completeness. The driver rejects retained coverage that advances beyond classified object work, root raw page indexes that outrun the fixture namespace, committed page coverage that exceeds indexed coverage, same-parent committed coverage regressions before completion, and any process-restart regression in retained coverage. No observed-name set, `readdir` offset, or assumed stable ordering is used as durable progress. Namespace creation happens only during fixture setup, before scan accounting.
 
-The `cfg(test)` hook observes actual entries delivered by `read_dir` and cancels the existing cycle token at the fixed entry limit. This is a deterministic injected **raw-entry work budget**, not a wall-clock performance measurement or a claim that kernel prefetch, probes, allocations, name bytes, or cache I/O are independently budgeted. The watchdog timeout only bounds worker lifetime. The hook does not replace enumeration, classification, or recursion, and does not exist in production builds. In particular, `xl.meta` object-boundary classification is unchanged.
+The worker sets `max_entries` on the production `ScannerCycleBudget`, and the `cfg(test)` hook only observes actual entries delivered by `read_dir`. This is a deterministic injected **raw-entry work budget**, not a wall-clock performance measurement or a claim that kernel prefetch, probes, allocations, or cache I/O are independently budgeted. Entries replayed up to the size of a validated index are read again on every restart and are bounded only by the runtime budget. The watchdog timeout only bounds worker lifetime. The hook does not replace enumeration, classification, or recursion, and does not exist in production builds. In particular, `xl.meta` object-boundary classification is unchanged.
 
-Exit 0 requires exact complete object/version/byte coverage within the same fixed budget on every executed round and positive evidence for all three stages: raw enumeration/indexing, object classification/processing, and durable cache retention after a fresh worker process reloads the previous report. Exit 1 means the strict convergence oracle remains unmet, including the current flat-directory enumeration starvation case. Exit 2 means invalid input, worker failure, or invalid evidence; it is not a successful reproduction. There is no final unbudgeted sweep. Small fixtures can pass; that does not establish the general R-E gate from [the scanner review comment](https://github.com/rustfs/backlog/issues/2240#issuecomment-5549222480). Raw entries observed are not a retained enumeration watermark. This is scanner-worker process restart plus codec evidence, **not** whole-daemon restart, EC quorum persistence, crash/fsync durability, remote RPC, or a throughput benchmark. The caller owns the bounded evidence directory and may remove it after inspection.
+Exit 0 requires exact complete object/version/byte coverage within the same fixed budget on every executed round and positive evidence for all three stages: raw enumeration/indexing, object classification/processing, and durable cache retention after a fresh worker process reloads the previous report. Exit 1 means the strict convergence oracle remains unmet. Exit 2 means invalid input, worker failure, or invalid evidence; it is not a successful reproduction. There is no final unbudgeted sweep. Small fixtures can pass; that does not establish the general R-E gate from [the scanner review comment](https://github.com/rustfs/backlog/issues/2240#issuecomment-5549222480). Raw entries observed are not a retained enumeration watermark. This is scanner-worker process restart plus codec evidence, **not** whole-daemon restart, EC quorum persistence, crash/fsync durability, remote RPC, or a throughput benchmark. The caller owns the bounded evidence directory and may remove it after inspection.
+
+### Budgeted First Build
+
+`FolderScanner::scan_folder` streams `read_dir`, charges each entry past the restored index size to the budget, and records it in the raw page index. Object metadata stays on the object budget. `max_entries` and `max_entry_name_bytes` stop a scan directly; `max_objects` and `max_directories` also bound one directory listing to 16 entries per budget unit, with a 128-entry floor, because a forward sweep needs the whole listing before it classifies in sorted order. A directory keeps its index until its children end, and the checkpoint saves every index on the active walk path, so a restart does not recharge any wide ancestor. A stop checkpoint grows only by the charged names. The total index stays proportional to the directory width.
+
+### Bucket Service Order
+
+Each set cache carries an optional `scan_bucket_service` map: a first-in-first-out wait position per bucket with its granted quanta, last certification and failure debt. A new or recreated bucket joins behind waiters but ahead of buckets that already had a turn in the current generation. The order is saved after budget stops and finished incomplete cycles too, under the set cache lock and revisions, and never when the owner was cancelled. It orders work and authorizes nothing, so another movement epoch or an invalid value starts a new order. Service recorded after the last save is not durable: a crash or cancellation inside a cycle repeats that cycle's order. Set dispatch order across restarts is not persisted.
+
+`cold_bucket_service_` tests fix N=17 buckets and q=4 service quanta with one hot and one failing bucket, and require service within ceil(17/4)=5 rounds through the real worker queue, the saved codec, and fresh OS processes. `fair_prefix_quantum_` tests inject small bucket shares. Production keeps `bucket_quantum` unset, so a bucket may use the total budget and then moves to the tail.
 
 ### Missing Storage Capability
 
-The current `scanner_folder::FolderScanner::scan_folder` collects child folders before recursing. `LocalDisk::scan_dir` also reads the whole parent before sorting and applying `forward_to`. Neither can supply a restartable first-build cursor over per-disk raw entries, orphan directories, and metadata boundaries. Repeated listing from the beginning is real work, not free pagination.
+`read_dir` has no restartable cursor, so resuming a listing rereads it from the start. Repeated listing from the beginning is real work, not free pagination.
 
-A future storage-owner capability must expose an explicit unsupported/building/ready state and a durable snapshot/index identity bound to disk mount, bucket incarnation, and directory identity. It must budget the first build and every page, including entry count, name bytes, metadata probes, I/O and time; survive a process restart during first build; seal page data before advancing the manifest; and distinguish enumerated, classified, and fully processed frontiers. An uncommitted page may be replayed only within a bounded cost. `xl.meta` classification must finish before descendants become traversable namespace. Missing capability or invalid identities must not become fabricated progress or completeness. No such capability is implemented by this diagnostic, and ordinary local storage remains without this R-E guarantee.
+A future storage-owner capability must expose an explicit unsupported/building/ready state and a durable snapshot/index identity bound to disk mount, bucket incarnation, and directory identity. It must budget the first build and every page, including entry count, name bytes, metadata probes, I/O and time; survive a process restart during first build; seal page data before advancing the manifest; and distinguish enumerated, classified, and fully processed frontiers. An uncommitted page may be replayed only within a bounded cost. `xl.meta` classification must finish before descendants become traversable namespace. Missing capability or invalid identities must not become fabricated progress or completeness. No such capability is implemented by this diagnostic. The page index above budgets new entries only, and ordinary local storage remains without this R-E guarantee.
 
 ## Completed Subtree Checkpoint Fixture