Merge branch 'main' into overtrue/scanner-pacing-batches

crates/scanner/src/scanner_folder.rs

@@ -34,7 +34,7 @@ use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetReason};
 use crate::scanner_io::{
     SCANNER_SKIP_FILE_ERROR, ScannerIODisk as _, is_scanner_metadata_corrupt_error, is_scanner_metadata_transient_error,
 };
-use crate::sleeper::DynamicSleeper;
+use crate::sleeper::{DynamicSleeper, PacingDebt};
 use crate::storage_api::owner::{
     EcstoreBucketLifecycleConfiguration as BucketLifecycleConfiguration, EcstoreEventArgs,
     EcstoreLifecycleRuleFilter as LifecycleRuleFilter, EcstoreObjectLockConfiguration as ObjectLockConfiguration,
@@ -769,6 +769,7 @@ pub struct FolderScanner {
     resolved_failed_object_paths: HashSet<String>,
 
     sleeper: DynamicSleeper,
+    pacing_debt: PacingDebt,
     // should_heal: Arc<dyn Fn() -> bool + Send + Sync>,
     disks: Vec<Arc<Disk>>,
     disks_quorum: usize,
@@ -1835,7 +1836,7 @@ impl FolderScanner {
                 // A candidate may be object payload rather than a namespace.
                 wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?;
             } else {
-                wait_for_scanner_io(&ctx, self.sleeper.sleep_folder()).await?;
+                wait_for_scanner_io(&ctx, self.sleeper.sleep_folder_paced(&mut self.pacing_debt)).await?;
             }
 
             let mut existing_folders: Vec<CachedFolder> = Vec::new();
@@ -2163,7 +2164,7 @@ impl FolderScanner {
                     continue;
                 }
 
-                let timer = self.sleeper.timer();
+                let work_started = tokio::time::Instant::now();
 
                 let heal_enabled = this_hash.mod_alt(
                     self.old_cache.info.next_cycle as u32 / folder.object_heal_prob_div,
@@ -2311,7 +2312,8 @@ impl FolderScanner {
                             }
                         }
 
-                        timer.sleep().await;
+                        wait_for_scanner_io(&ctx, self.sleeper.pace_object(&mut self.pacing_debt, work_started.elapsed()))
+                            .await?;
                         continue;
                     }
                 };
@@ -2334,7 +2336,7 @@ impl FolderScanner {
                 self.checkpoint_objects = self.checkpoint_objects.saturating_add(1);
                 self.maybe_send_checkpoint(&folder, &this_hash, into);
 
-                timer.sleep().await;
+                wait_for_scanner_io(&ctx, self.sleeper.pace_object(&mut self.pacing_debt, work_started.elapsed())).await?;
 
                 if ctx.is_cancelled() {
                     return Err(ScannerError::Other("Operation cancelled".to_string()));
@@ -3077,6 +3079,9 @@ impl FolderScanner {
             }
         }
 
+        if folder.parent.is_none() {
+            wait_for_scanner_io(&ctx, self.sleeper.flush_pacing(&mut self.pacing_debt)).await?;
+        }
         self.finish_size_reconciliation_batch();
         done_folder();
         let scanned_objects = u64::try_from(into.objects).unwrap_or(u64::MAX);
@@ -3197,6 +3202,7 @@ pub(crate) async fn scan_data_folder_scoped(
         failed_object_paths_seen: HashSet::new(),
         resolved_failed_object_paths: HashSet::new(),
         sleeper,
+        pacing_debt: PacingDebt::default(),
         disks,
         disks_quorum,
         updates,

crates/scanner/src/scanner_folder/tests.rs

@@ -18,6 +18,7 @@ use super::*;
 
 pub(super) mod erasure_probe;
 mod mrf_ownership;
+mod pacing;
 use crate::storage_api::VersionPurgeStatusType;
 use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
 use rustfs_filemeta::{FileInfo, FileMeta, MetadataResolutionParams};
@@ -364,6 +365,7 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
         failed_object_paths_seen: HashSet::new(),
         resolved_failed_object_paths: HashSet::new(),
         sleeper: SCANNER_SLEEPER.clone(),
+        pacing_debt: PacingDebt::default(),
         disks: Vec::new(),
         disks_quorum: 0,
         updates: None,

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

@@ -0,0 +1,148 @@
+// 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 crate::sleeper::SCANNER_IDLE_MODE;
+
+struct IdleGuard(bool);
+
+impl IdleGuard {
+    fn enabled() -> Self {
+        Self(SCANNER_IDLE_MODE.swap(true, Ordering::Relaxed))
+    }
+}
+
+impl Drop for IdleGuard {
+    fn drop(&mut self) {
+        SCANNER_IDLE_MODE.store(self.0, Ordering::Relaxed);
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn pacing_preserves_real_folder_versions_bytes_checkpoints_and_maintenance() {
+    let _idle = IdleGuard::enabled();
+    let mut totals = Vec::new();
+    let pending = pending_heal(PendingScannerHealKind::Object, "bucket", Some("unvisited"), None, 0, 1);
+    for factor in [0.0, 0.01] {
+        let (mut scanner, temp_dir) = build_test_scanner().await;
+        let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
+        scanner.old_cache.info.name = "bucket".to_string();
+        scanner.new_cache.info.name = "bucket".to_string();
+        scanner.update_cache.info.name = "bucket".to_string();
+        scanner.new_cache.info.pending_heals.push(pending.clone());
+        scanner.is_erasure_mode = true;
+        scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Default).for_disk_with_budget_for_tests(
+            &temp_dir,
+            crate::sleeper::ScannerDiskIoBudgetLimits {
+                iops: Some(1_000_000),
+                bytes_per_sec: Some(1_000_000),
+            },
+        );
+        scanner
+            .sleeper
+            .update_from_runtime_config(factor, Duration::from_secs(1), true, 64);
+        for object in 0..16 {
+            let name = format!("prefix/object-{object:03}");
+            let mut metadata = FileMeta::new();
+            for (version, size) in [(1, 3), (2, 5)] {
+                let mut info = FileInfo::new(&name, 4, 2);
+                info.volume = "bucket".to_string();
+                info.version_id = Some(Uuid::from_u128(version));
+                info.versioned = true;
+                info.size = size;
+                info.mod_time = Some(OffsetDateTime::from_unix_timestamp(1_700_000_000 + size).expect("fixture timestamp"));
+                metadata.add_version(info).expect("add metadata version");
+            }
+            write_test_object_metadata_bytes(&temp_dir, "bucket", &name, &metadata.marshal_msg().expect("encode versions")).await;
+        }
+        let mut entry = DataUsageEntry::default();
+        scanner
+            .scan_folder(
+                CancellationToken::new(),
+                CachedFolder {
+                    name: "bucket".to_string(),
+                    parent: None,
+                    object_heal_prob_div: 1,
+                },
+                &mut entry,
+            )
+            .await
+            .expect("complete real folder scan");
+        let total = scanner.new_cache.size_recursive(&hash_path("bucket").0).expect("root totals");
+        assert_eq!(total.objects, 16);
+        assert_eq!(total.versions, 32);
+        assert_eq!(total.size, 128);
+        assert_eq!(total.failed_objects, 0);
+        assert_eq!(scanner.checkpoint_objects, 16);
+        assert_eq!(scanner.new_cache.info.pending_heals, vec![pending.clone()]);
+        assert_eq!(scanner.pacing_debt.charged_objects, 16);
+        assert_eq!(scanner.pacing_debt.pending(), Duration::ZERO, "bucket completion settles the tail");
+        if factor > 0.0 {
+            assert!(scanner.pacing_debt.batches > 0, "the real object path must execute pacing");
+        }
+        let status = crate::sleeper::scanner_disk_io_budget_status()
+            .into_iter()
+            .find(|status| status.disk == temp_dir.to_string_lossy())
+            .expect("real walker disk charges");
+        assert!(status.charged_ops >= 16 * 2);
+        assert!(status.charged_bytes > 0);
+        totals.push((
+            serde_json::to_value(total).expect("serialize totals"),
+            status.charged_ops,
+            status.charged_bytes,
+        ));
+    }
+    assert_eq!(totals[0], totals[1], "batching must preserve full accounting and every disk charge");
+}
+
+#[tokio::test]
+#[serial]
+async fn pacing_settles_failed_metadata_work_at_bucket_completion() {
+    let _idle = IdleGuard::enabled();
+    let (mut scanner, temp_dir) = build_test_scanner().await;
+    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
+    scanner.old_cache.info.name = "bucket".to_string();
+    scanner.new_cache.info.name = "bucket".to_string();
+    scanner.update_cache.info.name = "bucket".to_string();
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Default);
+    scanner
+        .sleeper
+        .update_from_runtime_config(0.01, Duration::from_secs(1), true, 64);
+    write_test_object_metadata_bytes(&temp_dir, "bucket", "broken", &[]).await;
+    let mut entry = DataUsageEntry::default();
+    scanner
+        .scan_folder(
+            CancellationToken::new(),
+            CachedFolder {
+                name: "bucket".to_string(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut entry,
+        )
+        .await
+        .expect("corrupt metadata remains a recorded failure");
+    let total = scanner
+        .new_cache
+        .size_recursive(&hash_path("bucket").0)
+        .expect("failed root totals");
+    assert_eq!(total.objects, 0);
+    assert_eq!(total.failed_objects, 1);
+    assert_eq!(scanner.new_cache.info.failed_objects.len(), 1);
+    assert_eq!(scanner.checkpoint_objects, 0);
+    assert_eq!(scanner.pacing_debt.charged_objects, 1, "failure work must enter the same pacer");
+    assert_eq!(scanner.pacing_debt.batches, 1);
+    assert_eq!(scanner.pacing_debt.pending(), Duration::ZERO);
+}

crates/scanner/src/sleeper.rs

@@ -25,6 +25,9 @@ use rustfs_scanner_metrics::metrics::global_metrics;
 use tokio::time::Duration;
 
 mod io_budget;
+mod pacing;
+
+pub(crate) use pacing::PacingDebt;
 
 pub use io_budget::{
     SCANNER_DISK_IO_BUDGET_MAX_WAIT, ScannerDiskIoBudget, ScannerDiskIoBudgetConfig, ScannerDiskIoBudgetLimits,
@@ -131,6 +134,8 @@ pub static SCANNER_SLEEPER: LazyLock<DynamicSleeper> = LazyLock::new(|| {
 ///
 /// For operations timed via [`SleepTimer`], sleeps are computed as
 /// `elapsed_work_time * factor`, clamped to `[MIN_SLEEP, max_sleep]`.
+/// Folder walkers accumulate proportional object waits without the per-object
+/// floor, paying bounded batches and the final bucket tail.
 /// For folder-level gaps, [`sleep_folder`] uses `MIN_SLEEP * factor`,
 /// clamped only by `max_sleep`. All sleeps are gated on [`SCANNER_IDLE_MODE`];
 /// the per-disk background I/O budget bound with [`DynamicSleeper::for_disk`]
@@ -142,16 +147,27 @@ pub struct DynamicSleeper {
 }
 
 struct SleeperParams {
-    factor: RwLock<f64>,
-    max_sleep: RwLock<Duration>,
+    config: RwLock<PacingConfig>,
+    changed: tokio::sync::Notify,
+}
+
+#[derive(Clone, Copy)]
+struct PacingConfig {
+    factor: f64,
+    max_sleep: Duration,
+    reset_generation: u64,
 }
 
 impl DynamicSleeper {
     pub fn new(speed: ScannerSpeed) -> Self {
         Self {
             inner: Arc::new(SleeperParams {
-                factor: RwLock::new(speed.sleep_factor()),
-                max_sleep: RwLock::new(speed.max_sleep()),
+                config: RwLock::new(PacingConfig {
+                    factor: speed.sleep_factor(),
+                    max_sleep: speed.max_sleep(),
+                    reset_generation: 0,
+                }),
+                changed: tokio::sync::Notify::new(),
             }),
             disk_io_budget: None,
         }
@@ -213,9 +229,12 @@ impl DynamicSleeper {
     }
 
     fn read_params(&self) -> (f64, Duration) {
-        let factor = *self.inner.factor.read().unwrap_or_else(|e| e.into_inner());
-        let max_sleep = *self.inner.max_sleep.read().unwrap_or_else(|e| e.into_inner());
-        (factor, max_sleep)
+        let config = self.read_config();
+        (config.factor, config.max_sleep)
+    }
+
+    fn read_config(&self) -> PacingConfig {
+        *self.inner.config.read().unwrap_or_else(|e| e.into_inner())
     }
 
     /// Pace one directory listing: the preset folder gap, then one
@@ -260,10 +279,22 @@ impl DynamicSleeper {
     }
 
     fn update_params(&self, factor: f64, max_sleep: Duration) {
-        let mut f = self.inner.factor.write().unwrap_or_else(|e| e.into_inner());
-        *f = factor;
-        let mut m = self.inner.max_sleep.write().unwrap_or_else(|e| e.into_inner());
-        *m = max_sleep;
+        self.configure_pacing(factor, max_sleep, None);
+    }
+
+    fn configure_pacing(&self, factor: f64, max_sleep: Duration, idle_mode: Option<bool>) {
+        {
+            let mut config = self.inner.config.write().unwrap_or_else(|e| e.into_inner());
+            if let Some(idle_mode) = idle_mode {
+                SCANNER_IDLE_MODE.store(idle_mode, Ordering::Relaxed);
+            }
+            if factor == 0.0 || max_sleep.is_zero() || idle_mode == Some(false) {
+                config.reset_generation = config.reset_generation.wrapping_add(1);
+            }
+            config.factor = factor;
+            config.max_sleep = max_sleep;
+        }
+        self.inner.changed.notify_waiters();
     }
 
     /// Swap in parameters from a new speed preset (for runtime reconfiguration).
@@ -286,8 +317,7 @@ impl DynamicSleeper {
         idle_mode: bool,
         yield_every_n_objects: u64,
     ) {
-        self.update_params(sleep_factor, max_sleep);
-        SCANNER_IDLE_MODE.store(idle_mode, Ordering::Relaxed);
+        self.configure_pacing(sleep_factor, max_sleep, Some(idle_mode));
         self.record_throttle_config_with_yield(yield_every_n_objects);
     }
 

crates/scanner/src/sleeper/pacing.rs

@@ -0,0 +1,216 @@
+// 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 tokio::time::Instant;
+
+const PACING_QUANTUM: Duration = Duration::from_millis(1);
+const PACING_MAX_OBJECTS: u32 = 64;
+const PACING_MAX_AGE: Duration = Duration::from_millis(10);
+
+static PACING_OBJECTS: LazyLock<metrics::Counter> = LazyLock::new(|| {
+    metrics::describe_counter!(
+        "rustfs_scanner_pacing_objects_total",
+        "Object operations charged to walker proportional pacing; disk I/O is accounted separately."
+    );
+    metrics::counter!("rustfs_scanner_pacing_objects_total")
+});
+static PACING_BATCHES: LazyLock<metrics::Counter> = LazyLock::new(|| {
+    metrics::describe_counter!("rustfs_scanner_pacing_batches_total", "Walker proportional pacing batches executed.");
+    metrics::counter!("rustfs_scanner_pacing_batches_total")
+});
+
+/// Owned by one walker, including its recursive folder visits. No disk credit
+/// is stored here: cancelling this walker cannot refund already charged I/O.
+#[derive(Default)]
+pub(crate) struct PacingDebt {
+    pending: Duration,
+    objects: u32,
+    started: Option<Instant>,
+    reset_generation: u64,
+    #[cfg(test)]
+    pub(crate) charged_objects: u64,
+    #[cfg(test)]
+    pub(crate) batches: u64,
+}
+
+impl PacingDebt {
+    fn reset(&mut self) {
+        self.pending = Duration::ZERO;
+        self.objects = 0;
+        self.started = None;
+    }
+
+    fn prepare(&mut self, config: PacingConfig) {
+        if self.reset_generation != config.reset_generation
+            || !DynamicSleeper::should_sleep()
+            || config.factor == 0.0
+            || config.max_sleep.is_zero()
+        {
+            self.reset();
+        }
+        self.reset_generation = config.reset_generation;
+    }
+
+    fn charge(&mut self, delay: Duration, now: Instant) {
+        self.pending = self.pending.saturating_add(delay);
+        self.objects = self.objects.saturating_add(1);
+        if !self.pending.is_zero() {
+            self.started.get_or_insert(now);
+        }
+        PACING_OBJECTS.increment(1);
+        #[cfg(test)]
+        {
+            self.charged_objects += 1;
+        }
+    }
+
+    // Bounds are checked at scanner operation boundaries; an in-flight I/O
+    // operation does not create a second timer or task just to collect debt.
+    fn due(&self, config: PacingConfig, now: Instant) -> bool {
+        !self.pending.is_zero()
+            && (self.pending >= PACING_QUANTUM.min(config.max_sleep)
+                || self.objects >= PACING_MAX_OBJECTS
+                || self.started.is_some_and(|start| now.duration_since(start) >= PACING_MAX_AGE))
+    }
+
+    #[cfg(test)]
+    pub(crate) fn pending(&self) -> Duration {
+        self.pending
+    }
+}
+
+fn proportional_delay(elapsed: Duration, config: PacingConfig) -> Duration {
+    if !config.factor.is_finite() || config.factor <= 0.0 {
+        return Duration::ZERO;
+    }
+    Duration::try_from_secs_f64(elapsed.as_secs_f64() * config.factor)
+        .unwrap_or(config.max_sleep)
+        .min(config.max_sleep)
+}
+
+/// Dropping a cancelled wait still records the time actually spent waiting.
+struct PacingWait(Instant);
+
+impl Drop for PacingWait {
+    fn drop(&mut self) {
+        global_metrics().record_scanner_throttle_sleep(self.0.elapsed());
+    }
+}
+
+impl DynamicSleeper {
+    pub(crate) async fn pace_object(&self, debt: &mut PacingDebt, elapsed: Duration) {
+        self.pace(debt, Some(elapsed), false, false).await;
+    }
+
+    pub(crate) async fn sleep_folder_paced(&self, debt: &mut PacingDebt) {
+        self.pace(debt, None, true, false).await;
+        self.admit_disk_io(1, 0).await;
+    }
+
+    pub(crate) async fn flush_pacing(&self, debt: &mut PacingDebt) {
+        self.pace(debt, None, false, true).await;
+    }
+
+    async fn pace(&self, debt: &mut PacingDebt, elapsed: Option<Duration>, folder: bool, flush: bool) {
+        let config = self.read_config();
+        debt.prepare(config);
+        if !Self::should_sleep() || (flush && debt.pending.is_zero()) {
+            return;
+        }
+        if let Some(elapsed) = elapsed {
+            debt.charge(proportional_delay(elapsed, config), Instant::now());
+        }
+        let foreground = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+        let now = Instant::now();
+        let folder_reaches_deadline = folder
+            && debt.started.is_some_and(|start| {
+                now.duration_since(start)
+                    .saturating_add(proportional_delay(MIN_SLEEP, config))
+                    >= PACING_MAX_AGE
+            });
+        let repay = flush || !foreground.is_zero() || folder_reaches_deadline || debt.due(config, now);
+        if repay && !debt.pending.is_zero() {
+            PACING_BATCHES.increment(1);
+            #[cfg(test)]
+            {
+                debt.batches += 1;
+            }
+        }
+        if folder || repay || !foreground.is_zero() {
+            let mut folder = folder;
+            let mut foreground = foreground;
+            loop {
+                let requested = if repay { debt.pending } else { Duration::ZERO };
+                let paid = self.wait_pacing(requested, folder, foreground, debt.reset_generation).await;
+                if repay {
+                    debt.pending = debt.pending.saturating_sub(paid);
+                }
+                debt.prepare(self.read_config());
+                if !repay || debt.pending.is_zero() {
+                    break;
+                }
+                // A positive max_wait reduction caps each wait, not the debt.
+                // Actual timer overshoot also repays debt, so sub-ms caps do
+                // not require one timer per nanosecond of outstanding work.
+                folder = false;
+                foreground = Duration::ZERO;
+            }
+        }
+        if debt.pending.is_zero() {
+            debt.reset();
+        }
+    }
+
+    async fn wait_pacing(&self, debt: Duration, folder: bool, foreground: Duration, reset_generation: u64) -> Duration {
+        let start = Instant::now();
+        let mut measurement = None;
+        loop {
+            // Register before reading configuration so an update cannot be
+            // lost between choosing a duration and polling the notification.
+            let changed = self.inner.changed.notified();
+            tokio::pin!(changed);
+            changed.as_mut().enable();
+            let config = self.read_config();
+            if !Self::should_sleep() {
+                return Duration::ZERO;
+            }
+            let proportional = if config.reset_generation == reset_generation && config.factor > 0.0 {
+                debt.min(config.max_sleep)
+            } else {
+                Duration::ZERO
+            };
+            let folder_gap = if folder {
+                proportional_delay(MIN_SLEEP, config)
+            } else {
+                Duration::ZERO
+            };
+            let duration = proportional.saturating_add(folder_gap).max(foreground);
+            let remaining = duration.saturating_sub(start.elapsed());
+            if remaining.is_zero() {
+                return start.elapsed().saturating_sub(folder_gap);
+            }
+            measurement.get_or_insert_with(|| PacingWait(Instant::now()));
+            tokio::select! {
+                _ = tokio::time::sleep(remaining) => {
+                    return start.elapsed().saturating_sub(folder_gap);
+                }
+                _ = &mut changed => {}
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests;

crates/scanner/src/sleeper/pacing/tests.rs

@@ -0,0 +1,243 @@
+// 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 serial_test::serial;
+
+struct IdleGuard(bool);
+
+impl IdleGuard {
+    fn enabled() -> Self {
+        Self(SCANNER_IDLE_MODE.swap(true, Ordering::Relaxed))
+    }
+}
+
+impl Drop for IdleGuard {
+    fn drop(&mut self) {
+        SCANNER_IDLE_MODE.store(self.0, Ordering::Relaxed);
+    }
+}
+
+#[test]
+fn debt_boundaries_and_overflow() {
+    let config = DynamicSleeper::new(ScannerSpeed::Default).read_config();
+    let now = Instant::now();
+    let mut debt = PacingDebt::default();
+    debt.charge(Duration::ZERO, now);
+    assert!(!debt.due(config, now));
+    debt.charge(PACING_QUANTUM, now);
+    assert!(debt.due(config, now));
+    debt.reset();
+    debt.charge(PACING_QUANTUM + Duration::from_nanos(1), now);
+    assert!(debt.due(config, now));
+    debt.reset();
+    for _ in 0..PACING_MAX_OBJECTS - 1 {
+        debt.charge(Duration::from_nanos(1), now);
+    }
+    assert!(!debt.due(config, now));
+    assert!(debt.due(config, now + PACING_MAX_AGE));
+    debt.charge(Duration::from_nanos(1), now);
+    assert!(debt.due(config, now));
+    debt.charge(Duration::MAX, now);
+    debt.charge(Duration::MAX, now);
+    assert_eq!(debt.pending, Duration::MAX);
+    assert_eq!(proportional_delay(Duration::MAX, config), config.max_sleep);
+    assert_eq!(proportional_delay(Duration::ZERO, config), Duration::ZERO);
+    for factor in [f64::NAN, f64::INFINITY, -1.0] {
+        assert_eq!(
+            proportional_delay(Duration::from_secs(1), PacingConfig { factor, ..config }),
+            Duration::ZERO
+        );
+    }
+}
+
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn tiny_objects_conserve_debt_in_four_batches() {
+    let _idle = IdleGuard::enabled();
+    let sleeper = DynamicSleeper::new(ScannerSpeed::Default);
+    let mut debt = PacingDebt::default();
+    let start = Instant::now();
+    for _ in 0..100 {
+        sleeper.pace_object(&mut debt, Duration::from_micros(20)).await;
+    }
+    sleeper.flush_pacing(&mut debt).await;
+    assert_eq!(debt.charged_objects, 100);
+    assert_eq!(debt.batches, 4);
+    assert_eq!(start.elapsed(), Duration::from_millis(4));
+    assert_eq!(debt.pending, Duration::ZERO);
+}
+
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn short_bucket_tails_and_walkers_remain_independent() {
+    let _idle = IdleGuard::enabled();
+    let sleeper = DynamicSleeper::new(ScannerSpeed::Default);
+    let mut first = PacingDebt::default();
+    let mut second = PacingDebt::default();
+    sleeper.pace_object(&mut first, Duration::from_micros(20)).await;
+    sleeper.pace_object(&mut second, Duration::from_micros(20)).await;
+    assert_eq!(first.pending, Duration::from_micros(40));
+    assert_eq!(second.pending, Duration::from_micros(40));
+    sleeper.flush_pacing(&mut first).await;
+    assert_eq!(first.pending, Duration::ZERO);
+    assert_eq!(second.pending, Duration::from_micros(40));
+    sleeper.flush_pacing(&mut second).await;
+    for _ in 0..10 {
+        let mut debt = PacingDebt::default();
+        sleeper.pace_object(&mut debt, Duration::from_micros(20)).await;
+        sleeper.flush_pacing(&mut debt).await;
+        assert_eq!(debt.batches, 1);
+        assert_eq!(debt.pending, Duration::ZERO);
+    }
+}
+
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn aged_debt_is_repaid_at_the_next_folder_boundary() {
+    let _idle = IdleGuard::enabled();
+    let sleeper = DynamicSleeper::new(ScannerSpeed::Default);
+    let mut debt = PacingDebt::default();
+    sleeper.pace_object(&mut debt, Duration::from_micros(20)).await;
+    tokio::time::advance(PACING_MAX_AGE).await;
+    sleeper.sleep_folder_paced(&mut debt).await;
+    assert_eq!(debt.pending, Duration::ZERO);
+    assert_eq!(debt.batches, 1);
+}
+
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn positive_max_wait_reduction_preserves_pending_debt() {
+    let _idle = IdleGuard::enabled();
+    let sleeper = DynamicSleeper::new(ScannerSpeed::Default);
+    let mut debt = PacingDebt::default();
+    sleeper.pace_object(&mut debt, Duration::from_micros(400)).await;
+    sleeper.update_params(3.0, Duration::from_micros(100));
+    assert_eq!(debt.pending, Duration::from_micros(800));
+    let start = Instant::now();
+    sleeper.flush_pacing(&mut debt).await;
+    // Tokio timers have millisecond resolution; overshoot repays the debt.
+    assert!(start.elapsed() >= Duration::from_micros(800));
+    assert_eq!(debt.pending, Duration::ZERO);
+}
+
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn fastest_and_idle_off_clear_pending_debt_even_after_reenable() {
+    let _idle = IdleGuard::enabled();
+    let sleeper = DynamicSleeper::new(ScannerSpeed::Default);
+    for idle_off in [false, true] {
+        let mut debt = PacingDebt::default();
+        sleeper.pace_object(&mut debt, Duration::from_micros(20)).await;
+        if idle_off {
+            sleeper.update_from_runtime_config(2.0, Duration::from_secs(1), false, 64);
+        } else {
+            sleeper.update(ScannerSpeed::Fastest);
+        }
+        sleeper.update_from_runtime_config(2.0, Duration::from_secs(1), true, 64);
+        let start = Instant::now();
+        sleeper.flush_pacing(&mut debt).await;
+        assert_eq!(start.elapsed(), Duration::ZERO);
+        assert_eq!(debt.pending, Duration::ZERO);
+    }
+}
+
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn active_wait_wakes_for_fastest_idle_off_and_smaller_max_wait() {
+    let _idle = IdleGuard::enabled();
+    for update in 0..3 {
+        let sleeper = DynamicSleeper::new(ScannerSpeed::Slowest);
+        sleeper.update_from_runtime_config(100.0, Duration::from_secs(15), true, 64);
+        let mut debt = PacingDebt::default();
+        let start = Instant::now();
+        tokio::join!(sleeper.pace_object(&mut debt, Duration::from_secs(1)), async {
+            tokio::time::sleep(Duration::from_millis(5)).await;
+            match update {
+                0 => sleeper.update(ScannerSpeed::Fastest),
+                1 => sleeper.update_from_runtime_config(100.0, Duration::from_secs(15), false, 64),
+                _ => {
+                    sleeper.update_params(2.0, Duration::from_millis(1));
+                    // Positive updates preserve the old debt, so the wait
+                    // is still active until explicitly disabled.
+                    tokio::time::sleep(Duration::from_millis(5)).await;
+                    sleeper.update(ScannerSpeed::Fastest);
+                }
+            }
+        });
+        assert!(start.elapsed() <= Duration::from_millis(12));
+        assert_eq!(debt.pending, Duration::ZERO);
+    }
+}
+
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn foreground_backoff_flushes_tail_without_double_charging_or_max_wait_cap() {
+    let _idle = IdleGuard::enabled();
+    let sleeper = DynamicSleeper::new(ScannerSpeed::Default);
+    sleeper.update_params(2.0, Duration::from_micros(500));
+    let mut debt = PacingDebt::default();
+    sleeper.pace_object(&mut debt, Duration::from_micros(20)).await;
+    struct ForegroundGuard;
+    impl Drop for ForegroundGuard {
+        fn drop(&mut self) {
+            crate::reset_foreground_read_activity_for_test();
+        }
+    }
+    let _foreground = ForegroundGuard;
+    crate::set_foreground_read_activity(25);
+    let start = Instant::now();
+    sleeper.sleep_folder_paced(&mut debt).await;
+    assert_eq!(start.elapsed(), Duration::from_millis(250));
+    assert_eq!(debt.pending, Duration::ZERO);
+    assert_eq!(debt.batches, 1);
+}
+
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn cancellation_interrupts_long_pacing_without_refunding_disk_io() {
+    let _idle = IdleGuard::enabled();
+    let disk = tempfile::tempdir().expect("temporary budget identity");
+    let sleeper = DynamicSleeper::new(ScannerSpeed::Slowest).for_disk_with_budget_for_tests(
+        disk.path(),
+        ScannerDiskIoBudgetLimits {
+            iops: Some(1),
+            bytes_per_sec: Some(1000),
+        },
+    );
+    sleeper.admit_disk_io(1, 100).await;
+    let ctx = tokio_util::sync::CancellationToken::new();
+    let mut debt = PacingDebt::default();
+    let start = Instant::now();
+    tokio::join!(
+        async {
+            tokio::select! {
+                biased;
+                _ = ctx.cancelled() => {}
+                _ = sleeper.pace_object(&mut debt, Duration::from_secs(1)) => panic!("cancel must interrupt the wait"),
+            }
+        },
+        async {
+            tokio::time::sleep(Duration::from_millis(5)).await;
+            ctx.cancel();
+        }
+    );
+    assert_eq!(start.elapsed(), Duration::from_millis(5));
+    let status = scanner_disk_io_budget_status()
+        .into_iter()
+        .find(|s| s.disk == disk.path().to_string_lossy())
+        .expect("disk status");
+    assert_eq!(status.charged_ops, 1);
+    assert_eq!(status.charged_bytes, 100);
+}