fix(scanner): honor explicit cycle cadence (#6313)

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>

crates/scanner/src/scanner.rs

@@ -1804,14 +1804,13 @@ async fn run_data_scanner_with_maintenance_state(
             wait_plan.delay,
             activity_poll_interval,
             &mut scanner_activity_seen,
-            ScannerCycleObservedGenerations {
-                // A non-converged cycle holds further activity notifications
-                // until its bounded retry timer to avoid an unbroken scan loop.
-                dirty_usage: convergence_retry_interval.is_none().then_some(dirty_usage_generation_seen),
-                runtime_config: runtime_config_generation_seen,
-                maintenance: maintenance_generation_before_wait,
-                defer_cluster_activity: convergence_retry_interval.is_some(),
-            },
+            ScannerCycleObservedGenerations::for_wait(
+                &runtime_config,
+                convergence_retry_interval,
+                dirty_usage_generation_seen,
+                runtime_config_generation_seen,
+                maintenance_generation_before_wait,
+            ),
             || guard.is_lock_lost(),
             || probe_scanner_activity(storeapi.as_ref(), distributed),
         )

crates/scanner/src/scanner/activity.rs

@@ -229,6 +229,27 @@ pub(super) struct ScannerCycleObservedGenerations {
     pub(super) defer_cluster_activity: bool,
 }
 
+impl ScannerCycleObservedGenerations {
+    pub(super) fn for_wait(
+        runtime_config: &ScannerRuntimeConfig,
+        convergence_retry_interval: Option<Duration>,
+        dirty_usage_generation_seen: u64,
+        runtime_config_generation: u64,
+        maintenance_generation: u64,
+    ) -> Self {
+        Self {
+            // An explicit cycle override is a duty-cycle policy; dirty usage
+            // wakes stay on the default adaptive path so the interval holds.
+            dirty_usage: (convergence_retry_interval.is_none()
+                && runtime_config.cycle_interval_source == ScannerRuntimeConfigSource::Default)
+                .then_some(dirty_usage_generation_seen),
+            runtime_config: runtime_config_generation,
+            maintenance: maintenance_generation,
+            defer_cluster_activity: convergence_retry_interval.is_some(),
+        }
+    }
+}
+
 pub(super) const LOCAL_SCANNER_ACTIVITY_NODE: &str = "<local>";
 
 #[derive(Clone, Debug, PartialEq, Eq)]

crates/scanner/src/scanner/tests.rs

@@ -3107,6 +3107,31 @@ fn clean_idle_backoff_requires_activity_probes() {
     assert!(!scanner_activity_probe_required(true, false, lifecycle, &default_config));
 }
 
+#[test]
+fn dirty_usage_wakes_are_disabled_for_explicit_cycle_policy() {
+    let default_config = ScannerRuntimeConfig::default();
+
+    let default_observed = ScannerCycleObservedGenerations::for_wait(&default_config, None, 7, 11, 13);
+    assert_eq!(default_observed.dirty_usage, Some(7));
+    assert_eq!(default_observed.runtime_config, 11);
+    assert_eq!(default_observed.maintenance, 13);
+    assert!(!default_observed.defer_cluster_activity);
+
+    let retry_observed = ScannerCycleObservedGenerations::for_wait(&default_config, Some(Duration::from_secs(11)), 7, 11, 13);
+    assert_eq!(retry_observed.dirty_usage, None);
+    assert!(retry_observed.defer_cluster_activity);
+
+    for source in [ScannerRuntimeConfigSource::Env, ScannerRuntimeConfigSource::Config] {
+        let explicit_cycle = ScannerRuntimeConfig {
+            cycle_interval_source: source,
+            ..default_config.clone()
+        };
+        let explicit_observed = ScannerCycleObservedGenerations::for_wait(&explicit_cycle, None, 7, 11, 13);
+        assert_eq!(explicit_observed.dirty_usage, None);
+        assert!(!explicit_observed.defer_cluster_activity);
+    }
+}
+
 #[test]
 #[serial]
 fn clean_idle_cap_preserves_default_bitrot_coverage_window() {