test: complete mixed heal fixture replicas before fault injection

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -376,6 +376,21 @@ struct WalkCopyError {
     source: io::Error,
 }
 
+impl WalkCopyError {
+    fn into_disk_error(self) -> DiskError {
+        // Only a local output write/flush proves the metacache consumer closed.
+        // An identical I/O kind from the remote input remains a transport failure.
+        if matches!(self.reason, RemoteWalkFailureReason::Write | RemoteWalkFailureReason::Flush)
+            && self.source.kind() == io::ErrorKind::BrokenPipe
+            && self.source.get_ref().is_none()
+        {
+            DiskError::metacache_output_stream_closed()
+        } else {
+            DiskError::Io(self.source)
+        }
+    }
+}
+
 async fn copy_stream_with_buffer<R, W>(
     reader: &mut R,
     writer: &mut W,
@@ -2096,6 +2111,7 @@ impl RemoteDisk {
         failure_health_action: FailureHealthAction,
     ) {
         if let Err(err) = operation_result
+            && !err.is_metacache_output_stream_closed()
             && is_network_like_disk_error(err)
         {
             counter!(
@@ -4056,7 +4072,7 @@ impl DiskAPI for RemoteDisk {
                             Ok(_) => return Ok(()),
                             Err(error) => {
                                 failure_reason = Some(error.reason);
-                                return Err(DiskError::Io(error.source));
+                                return Err(error.into_disk_error());
                             }
                         }
                     }
@@ -9687,6 +9703,43 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn remote_walk_preserves_local_metacache_consumer_close() {
+        let transport =
+            RetryingWalkDirInternodeDataTransport::with_steps(vec![WalkDirTestStep::Data(b"metacache payload".to_vec())]);
+        let disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await;
+        let (mut writer, reader) = tokio::io::duplex(1);
+        drop(reader);
+        let error = disk
+            .walk_dir(WalkDirOptions::default(), &mut writer)
+            .await
+            .expect_err("a dropped metacache consumer must stop the producer");
+        assert!(
+            error.is_metacache_output_stream_closed(),
+            "preserve the benign completion marker: {error:?}"
+        );
+        assert_eq!(transport.calls().len(), 1, "local consumer closure is not a remote open retry");
+        assert!(!disk.health.is_faulty());
+    }
+
+    #[tokio::test]
+    async fn remote_walk_keeps_remote_read_broken_pipe_as_transport_failure() {
+        let transport = RetryingWalkDirInternodeDataTransport::with_steps(vec![WalkDirTestStep::PartialDataThenError {
+            data: b"partial metacache".to_vec(),
+            error: io::ErrorKind::BrokenPipe.into(),
+        }]);
+        let disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await;
+        let mut writer = Vec::new();
+        let error = disk
+            .walk_dir(WalkDirOptions::default(), &mut writer)
+            .await
+            .expect_err("an upstream close after partial output is a failed walk");
+        assert!(!error.is_metacache_output_stream_closed());
+        assert!(error.contains_io_error_kind(io::ErrorKind::BrokenPipe));
+        assert_eq!(writer, b"partial metacache");
+        assert_eq!(transport.calls().len(), 1, "partial output must not be replayed");
+    }
+
     #[tokio::test]
     async fn walk_stream_backpressure_limits_read_ahead_and_is_cancellable() {
         let mut reader = Cursor::new(vec![5; 4 * DEFAULT_READ_BUFFER_SIZE]);

crates/ecstore/src/diagnostics/admin_server_info.rs

@@ -682,15 +682,11 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32,
         return Err(Error::other("ServerNotInitialized"));
     };
 
-    let mut pools_info: HashMap<i32, HashMap<i32, ErasureSetInfo>> = HashMap::new();
-    for d in all_disks {
-        let pool_info = pools_info.entry(d.pool_index).or_default();
-        let erasure_set = pool_info.entry(d.set_index).or_default();
-
-        if erasure_set.id == 0 {
-            erasure_set.id = d.set_index;
+    collect_pools_info(all_disks, |pool_index, set_index| {
+        let store = store.clone();
+        async move {
             let load = load_data_usage_cache(
-                store.pools[d.pool_index as usize].disk_set[d.set_index as usize].as_ref(),
+                store.pools[pool_index as usize].disk_set[set_index as usize].as_ref(),
                 DATA_USAGE_CACHE_NAME,
             );
             #[cfg(feature = "diagnostic-read-version-caller")]
@@ -698,11 +694,34 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32,
                 rustfs_io_metrics::internode_metrics::ReadVersionEntrySource::ServerInfoPools,
                 load,
             );
-            match load.await {
-                Ok(cache) => apply_erasure_set_usage(&cache, erasure_set),
-                Err(_) => erasure_set.usage_error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()),
-            }
+            load.await
         }
+    })
+    .await
+}
+
+async fn collect_pools_info<F, Fut>(all_disks: &[Disk], mut load_usage: F) -> Result<HashMap<i32, HashMap<i32, ErasureSetInfo>>>
+where
+    F: FnMut(i32, i32) -> Fut,
+    Fut: std::future::Future<Output = Result<rustfs_data_usage::DataUsageCache>>,
+{
+    let mut pools_info: HashMap<i32, HashMap<i32, ErasureSetInfo>> = HashMap::new();
+    for d in all_disks {
+        let pool_info = pools_info.entry(d.pool_index).or_default();
+        let erasure_set = match pool_info.entry(d.set_index) {
+            std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(),
+            std::collections::hash_map::Entry::Vacant(entry) => {
+                let erasure_set = entry.insert(ErasureSetInfo {
+                    id: d.set_index,
+                    ..Default::default()
+                });
+                match load_usage(d.pool_index, d.set_index).await {
+                    Ok(cache) => apply_erasure_set_usage(&cache, erasure_set),
+                    Err(_) => erasure_set.usage_error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()),
+                }
+                erasure_set
+            }
+        };
 
         erasure_set.raw_capacity += d.total_space;
         erasure_set.raw_usage += d.used_space;
@@ -1241,6 +1260,59 @@ mod tests {
         assert_eq!(info.deployment_id, expected_deployment_id);
     }
 
+    #[tokio::test]
+    async fn pool_usage_is_loaded_once_per_set_including_set_zero() {
+        let mut disks = Vec::new();
+        for (pool, set, count) in [(0, 0, 16), (1, 0, 16), (1, 1, 4)] {
+            disks.extend((0..count).map(|_| Disk {
+                pool_index: pool,
+                set_index: set,
+                used_space: 10,
+                total_space: 100,
+                ..Default::default()
+            }));
+        }
+        let mut loads = Vec::new();
+        let info = super::collect_pools_info(&disks, |pool, set| {
+            loads.push((pool, set));
+            async move {
+                if pool == 1 && set == 0 {
+                    return Err(crate::error::Error::other("usage unavailable"));
+                }
+                let mut cache = rustfs_data_usage::DataUsageCache::default();
+                cache.info.name = DATA_USAGE_ROOT.to_owned();
+                cache.info.last_update = Some(std::time::SystemTime::UNIX_EPOCH);
+                cache.info.snapshot_complete = true;
+                cache.replace(
+                    DATA_USAGE_ROOT,
+                    "",
+                    rustfs_data_usage::DataUsageEntry {
+                        size: 25,
+                        objects: 2,
+                        ..Default::default()
+                    },
+                );
+                Ok(cache)
+            }
+        })
+        .await
+        .expect("collect mixed successful and unavailable usage");
+        assert_eq!(loads, [(0, 0), (1, 0), (1, 1)]);
+        for (pool, set, count) in [(0, 0, 16), (1, 0, 16), (1, 1, 4)] {
+            let set_info = &info[&pool][&set];
+            assert_eq!(set_info.id, set);
+            assert_eq!(set_info.raw_usage, count * 10);
+            assert_eq!(set_info.raw_capacity, count * 100);
+            if pool == 1 && set == 0 {
+                assert_eq!(set_info.usage_error.as_deref(), Some(DATA_USAGE_UNAVAILABLE_ERROR));
+            } else {
+                assert_eq!(set_info.usage, 25, "logical usage must not multiply by drive count");
+                assert_eq!(set_info.objects_count, 2);
+                assert!(set_info.usage_error.is_none());
+            }
+        }
+    }
+
     #[serial]
     #[tokio::test]
     async fn local_server_property_includes_runtime_stats_without_endpoint_pools() {

crates/ecstore/src/set_disk/read.rs

@@ -2979,10 +2979,11 @@ fn build_legacy_per_part_fallback_reader(
     part_length: usize,
     part_size: usize,
     read_repair: Option<(DecodeReadRepairContext, bool)>,
-) -> Box<dyn AsyncRead + Unpin + Send + Sync> {
+) -> Box<LegacyPerPartDecodeReader> {
     let buffer = adaptive_duplex_buffer_size(part_size as i64);
-    let (read_half, mut write_half) = tokio::io::duplex(buffer);
+    let (read_half, write_half) = tokio::io::duplex(buffer);
     let decode = tokio::spawn(async move {
+        let mut write_half = GetObjectDownstreamWriter::new(write_half);
         let outcome = erasure
             .decode_with_stripe_handles_and_reopeners_with_diagnostics(
                 &mut write_half,
@@ -8908,6 +8909,42 @@ mod tests {
         assert!(err.to_string().contains("inconsistent read source shards"));
     }
 
+    #[tokio::test]
+    async fn legacy_per_part_output_close_retains_downstream_classification() {
+        let erasure = coding::Erasure::new(2, 2, 64);
+        let data = (0..64u8).collect::<Vec<_>>();
+        let setup =
+            setup_codec_data_blocks_first_encoded_bitrot_readers(&erasure, &data, &[0], &[], &[], HashAlgorithm::HighwayHash256)
+                .await;
+        let mut reader = build_legacy_per_part_fallback_reader(
+            erasure,
+            setup.readers,
+            setup.deferred_stripe_handles,
+            Vec::new(),
+            None,
+            0,
+            data.len(),
+            data.len(),
+            None,
+        );
+        // Closing the actual output pipe models the body consumer disappearing
+        // while retaining the executor long enough to inspect its typed result.
+        assert!(!reader.finished);
+        let (replacement, _unused) = tokio::io::duplex(1);
+        drop(std::mem::replace(&mut reader.inner, replacement));
+        let handle = reader.decode.take().expect("fallback decode task");
+        let error = handle
+            .await
+            .expect("join the fallback task")
+            .expect("closed pipe must interrupt output");
+        assert_eq!(error.kind(), std::io::ErrorKind::BrokenPipe);
+        assert_eq!(
+            crate::diagnostics::get::classify_io_error(&error),
+            GetObjectFailureReason::DownstreamClosed,
+            "fallback output must carry the same marker as the normal GET path"
+        );
+    }
+
     // backlog#879: a later multipart part that loses a shard (read quorum still
     // holds) must be reconstructed in place through the legacy per-part decode
     // bridge instead of failing the stream. This exercises the exact reader the

crates/heal/tests/shard_identity_receipt_test.rs

@@ -30,6 +30,78 @@ use storage_api::integration::{
     WriteCommitGuard, WriteCompletion,
 };
 
+async fn complete_fixture_replicas(env: &TestECStoreEnv, bucket: &str, object: &str) {
+    // TailDrained completes fan-out, but a successful quorum write may still
+    // have minority failures. Fault injection needs a fully repaired baseline.
+    let result = ECStoreHealStorage::new(env.ecstore.clone())
+        .heal_object_with_receipt(
+            bucket,
+            object,
+            None,
+            &HealOpts {
+                scan_mode: HealScanMode::Deep,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("complete protected fixture replicas before fault injection");
+    assert!(result.error.is_none(), "fixture repair failed: {:?}", result.error);
+    assert_eq!(result.item.after.drives.len(), env.disk_paths.len());
+    assert!(
+        result.item.after.drives.iter().all(|drive| drive.state == "ok"),
+        "fixture requires every replica, not only read/write quorum: {:?}",
+        result.item.after.drives
+    );
+    assert!(result.item.integrity_verified, "fixture must retain protected payload verification");
+}
+
+#[tokio::test]
+#[serial]
+async fn mixed_batch_fixture_repairs_a_missing_minority_replica() {
+    let root = tempfile::tempdir().expect("minority replica fixture");
+    let env = TestECStoreEnv::builder().base_dir(root.path()).build().await;
+    let bucket = "mixed-fixture-baseline";
+    let object = "object";
+    env.make_bucket(bucket, false).await;
+    let expected = vec![0x5a; 1024 * 1024 + 123];
+    env.ecstore
+        .put_object(
+            bucket,
+            object,
+            &mut PutObjReader::from_vec(expected.clone()),
+            &ObjectOptions {
+                shard_integrity_write_mode: Some(ShardIntegrityWriteMode::Protected),
+                write_completion: WriteCompletion::TailDrained,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("quorum commit fixture");
+    // Guarantee a minority failure regardless of whether the original write
+    // succeeded on every disk, then repair it before any test fault is applied.
+    let replica = env.disk_paths[0].join(bucket).join(object);
+    if replica.exists() {
+        tokio::fs::remove_dir_all(&replica)
+            .await
+            .expect("remove one complete replica");
+    }
+    complete_fixture_replicas(&env, bucket, object).await;
+    let disks = env.ecstore.pools[0].get_disks(0);
+    for disk in disks.disks.read().await.iter().flatten() {
+        disk.read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("every fixture replica has metadata after completion");
+    }
+    let mut reader = env
+        .ecstore
+        .get_object_reader(bucket, object, None, Default::default(), &ObjectOptions::default())
+        .await
+        .expect("read the completed fixture");
+    let mut actual = Vec::new();
+    reader.stream.read_to_end(&mut actual).await.expect("completed fixture body");
+    assert_eq!(actual, expected);
+}
+
 #[test]
 fn disabled_lock_reports_only_quorum_selection_and_deep_check_observations() {
     const CHILD_ENV: &str = "RUSTFS_TEST_DISABLED_LOCK_HEAL_CHILD";
@@ -704,6 +776,7 @@ fn admin_mixed_batch_retains_repairs_after_object_retry_exhaustion() {
                             )
                             .await
                             .expect("commit complete fixture");
+                        complete_fixture_replicas(&env, bucket, object).await;
                         for (slot, disk) in disks.iter().enumerate() {
                             let meta = disk
                                 .read_version("", bucket, object, "", &ReadOptions::default())

crates/scanner/src/sleeper.rs

@@ -230,6 +230,17 @@ impl DynamicSleeper {
         SCANNER_IDLE_MODE.load(Ordering::Relaxed)
     }
 
+    fn foreground_backoff(&self) -> Duration {
+        // Local walkers already use disk-priority admission and a shared I/O
+        // budget. A global per-operation floor duplicates that control and
+        // throttles idle disks because unrelated foreground work is active.
+        if self.disk_io_budget.is_some() {
+            Duration::ZERO
+        } else {
+            foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity())
+        }
+    }
+
     fn read_params(&self) -> (f64, Duration) {
         let config = self.read_config();
         (config.factor, config.max_sleep)
@@ -252,15 +263,14 @@ impl DynamicSleeper {
         }
         let (factor, max_sleep) = self.read_params();
         if factor == 0.0 || max_sleep.is_zero() {
-            let foreground_sleep =
-                foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+            let foreground_sleep = self.foreground_backoff();
             if !foreground_sleep.is_zero() {
                 tokio::time::sleep(foreground_sleep).await;
                 global_metrics().record_scanner_throttle_sleep(foreground_sleep);
             }
             return;
         }
-        let foreground_sleep = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+        let foreground_sleep = self.foreground_backoff();
         let sleep_dur = Duration::from_secs_f64(MIN_SLEEP.as_secs_f64() * factor)
             .min(max_sleep)
             .max(foreground_sleep);
@@ -354,8 +364,7 @@ impl SleepTimer {
         }
         let (factor, max_sleep) = self.sleeper.read_params();
         if factor == 0.0 || max_sleep.is_zero() {
-            let foreground_sleep =
-                foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+            let foreground_sleep = self.sleeper.foreground_backoff();
             if !foreground_sleep.is_zero() {
                 tokio::time::sleep(foreground_sleep).await;
                 global_metrics().record_scanner_throttle_sleep(foreground_sleep);
@@ -363,7 +372,7 @@ impl SleepTimer {
             return;
         }
         let elapsed = self.start.elapsed();
-        let foreground_sleep = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+        let foreground_sleep = self.sleeper.foreground_backoff();
         let sleep_dur = Duration::from_secs_f64(elapsed.as_secs_f64() * factor)
             .max(MIN_SLEEP)
             .min(max_sleep)

crates/scanner/src/sleeper/pacing.rs

@@ -135,7 +135,7 @@ impl DynamicSleeper {
         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 foreground = self.foreground_backoff();
         let now = Instant::now();
         let folder_reaches_deadline = folder
             && debt.started.is_some_and(|start| {

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

@@ -181,6 +181,54 @@ async fn active_wait_wakes_for_fastest_idle_off_and_smaller_max_wait() {
     }
 }
 
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn disk_bound_pacing_does_not_repeat_global_foreground_backoff() {
+    let _idle = IdleGuard::enabled();
+    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 disk = tempfile::tempdir().expect("disk identity");
+    let sleeper = DynamicSleeper::new(ScannerSpeed::Default).for_disk_with_budget_for_tests(
+        disk.path(),
+        ScannerDiskIoBudgetLimits {
+            iops: Some(0),
+            bytes_per_sec: None,
+        },
+    );
+    let mut debt = PacingDebt::default();
+    let begin = Instant::now();
+    for _ in 0..100 {
+        sleeper.pace_object(&mut debt, Duration::from_micros(20)).await;
+    }
+    sleeper.flush_pacing(&mut debt).await;
+    assert_eq!(begin.elapsed(), Duration::from_millis(4));
+    assert_eq!(debt.charged_objects, 100);
+    assert_eq!(debt.pending, Duration::ZERO);
+    // The first operation consumes the token bucket's one-operation burst.
+    sleeper.admit_disk_io(1, 0).await;
+    let begin = Instant::now();
+    sleeper.sleep_folder_paced(&mut debt).await;
+    assert_eq!(
+        begin.elapsed(),
+        SCANNER_DISK_IO_BUDGET_MAX_WAIT + Duration::from_millis(2),
+        "the per-disk budget and preset folder gap must still apply"
+    );
+    assert_eq!(
+        scanner_disk_io_budget_status()
+            .into_iter()
+            .find(|status| status.disk == disk.path().to_string_lossy())
+            .expect("disk charges remain visible")
+            .charged_ops,
+        2
+    );
+}
+
 #[tokio::test(start_paused = true)]
 #[serial]
 async fn foreground_backoff_flushes_tail_without_double_charging_or_max_wait_cap() {

docs/operations/scanner-runtime-controls.md

@@ -40,7 +40,7 @@ Bitrot cycle resolution differs because the canonical persistent key belongs to
 | `scanner.cycle_max_objects` | `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` | objects | `0` (`DEFAULT_SCANNER_CYCLE_MAX_OBJECTS`) | Caps objects processed by one cycle. `0` disables this budget. |
 | `scanner.cycle_max_directories` | `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` | directories | `0` (`DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES`) | Caps directories entered by one cycle. `0` disables this budget. |
 | `heal.bitrot_cycle` | `RUSTFS_SCANNER_BITROT_CYCLE_SECS` | seconds | `2592000` (`DEFAULT_HEAL_BITROT_CYCLE_SECS`, 30 days) | Controls periodic deep bitrot scans. `false`, `off`, `no`, or `disabled` disables periodic deep scans; `0`, `true`, `on`, or `yes` runs deep mode every scanner cycle. |
-| `scanner.idle_mode` | `RUSTFS_SCANNER_IDLE_MODE` | boolean | `true` (`DEFAULT_SCANNER_IDLE_MODE`) | Enables preset scanner sleeps and foreground-read backoff. `false` disables those sleeps; per-disk foreground-priority admission remains active. |
+| `scanner.idle_mode` | `RUSTFS_SCANNER_IDLE_MODE` | boolean | `true` (`DEFAULT_SCANNER_IDLE_MODE`) | Enables preset scanner sleeps and the foreground backoff for callers without a disk binding. `false` disables those sleeps; per-disk foreground-priority admission remains active. |
 | `scanner.cache_save_timeout` | `RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS` | seconds | `14` (`DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS`) | Timeout for saving scanner cache; runtime enforces a minimum of `1` and keeps the default persistence budget within the distributed publication lease. |
 | `scanner.recovery_backoff_cap` | `RUSTFS_SCANNER_RECOVERY_BACKOFF_CAP_SECS` | seconds | `600` (`DEFAULT_SCANNER_RECOVERY_BACKOFF_CAP_SECS`) | Caps the wait between automatic startup recovery retries. Must be a positive integer; `0` and malformed values are rejected, so retries never stop. Applies to the next computed retry, not to a wait already in progress. The retry count resets after 24 hours, which bounds a single wait at about one day even with a larger cap. |
 | `scanner.max_concurrent_set_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS`) | Caps concurrent set-level scanner tasks. `0` keeps topology-derived concurrency. |
@@ -63,7 +63,7 @@ Speed presets (`crates/config/src/constants/scanner.rs`) set the base sleep mult
 | `slow` | 10x | 15s | 60s |
 | `slowest` | 100x | 15s | 30m |
 
-Use `scanner.delay`, `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis needs a precise override. With `idle_mode=true`, directory-level sleep is `1ms x factor` and object-level sleep is `time spent on the object x factor`, both capped at `max_wait`; a foreground-read floor of `FOREGROUND_READ_BACKOFF_PER_REQUEST_MS` (10ms) per concurrent GetObject/streaming read, capped at `FOREGROUND_READ_BACKOFF_MAX_MS` (250ms), is applied on top and can exceed the preset's `max_wait` (`crates/scanner/src/sleeper.rs`). The per-disk foreground-priority gate is independent of `idle_mode` and remains active when sleeps are disabled.
+Use `scanner.delay`, `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis needs a precise override. With `idle_mode=true`, directory-level sleep is `1ms x factor` and object-level sleep is admitted work time multiplied by the factor, capped at `max_wait`; object pacing batches small waits and settles remaining debt at bucket completion. Local walkers bound with `DynamicSleeper::for_disk` use disk-priority admission and their shared I/O budget without an additional global foreground-request sleep. A caller without that binding retains a foreground-workload floor of 10ms per active request, capped at 250ms, which can exceed `max_wait` (`crates/scanner/src/sleeper.rs`). The per-disk foreground-priority gate and disk budget remain active when preset sleeps are disabled. Removing redundant global backoff does not guarantee a fixed snapshot age: scans still compete for admitted resources and may checkpoint at their cycle budget.
 
 ### Startup and cycle budgets