mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-19 11:06:17 +00:00
fix(scanner): respect configured scan start delay (#2119)
This commit is contained in:
@@ -35,8 +35,6 @@ use tokio::time::{Duration, Instant};
|
|||||||
use tokio_util::sync::CancellationToken;
|
use tokio_util::sync::CancellationToken;
|
||||||
use tracing::{debug, error, info, instrument, warn};
|
use tracing::{debug, error, info, instrument, warn};
|
||||||
|
|
||||||
const LOCK_RETRY_MAX: Duration = Duration::from_secs(30);
|
|
||||||
|
|
||||||
/// Returns the base cycle interval. If `RUSTFS_DATA_SCANNER_START_DELAY_SECS`
|
/// Returns the base cycle interval. If `RUSTFS_DATA_SCANNER_START_DELAY_SECS`
|
||||||
/// is set, it takes precedence; otherwise the value is derived from the
|
/// is set, it takes precedence; otherwise the value is derived from the
|
||||||
/// `RUSTFS_SCANNER_SPEED` preset.
|
/// `RUSTFS_SCANNER_SPEED` preset.
|
||||||
@@ -51,7 +49,11 @@ fn cycle_interval() -> Duration {
|
|||||||
/// Compute a randomized inter-cycle sleep.
|
/// Compute a randomized inter-cycle sleep.
|
||||||
// Delay is scan interval +- 10%, with a floor of 1 second.
|
// Delay is scan interval +- 10%, with a floor of 1 second.
|
||||||
fn randomized_cycle_delay() -> Duration {
|
fn randomized_cycle_delay() -> Duration {
|
||||||
let interval = cycle_interval().max(Duration::from_secs(1));
|
randomized_cycle_delay_for(cycle_interval())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn randomized_cycle_delay_for(interval: Duration) -> Duration {
|
||||||
|
let interval = interval.max(Duration::from_secs(1));
|
||||||
// Uniform in [-0.1, 0.1), keeping actual delay within 10% of interval.
|
// Uniform in [-0.1, 0.1), keeping actual delay within 10% of interval.
|
||||||
let jitter_factor = (rand::random::<f64>() * 0.2) - 0.1;
|
let jitter_factor = (rand::random::<f64>() * 0.2) - 0.1;
|
||||||
let delay = interval.mul_f64(1.0 + jitter_factor);
|
let delay = interval.mul_f64(1.0 + jitter_factor);
|
||||||
@@ -76,8 +78,12 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
|
|||||||
if let Err(e) = run_data_scanner(ctx_clone.clone(), storeapi_clone.clone()).await {
|
if let Err(e) = run_data_scanner(ctx_clone.clone(), storeapi_clone.clone()).await {
|
||||||
error!("Failed to run data scanner: {e}");
|
error!("Failed to run data scanner: {e}");
|
||||||
}
|
}
|
||||||
// Sleep if couldn't acquire lock or scan failed
|
// Backoff before retrying after lock contention or scanner-level failures.
|
||||||
tokio::time::sleep(randomized_cycle_delay().min(LOCK_RETRY_MAX)).await;
|
// Keep this cancellation-aware so shutdown is not delayed by backoff sleep.
|
||||||
|
tokio::select! {
|
||||||
|
_ = ctx_clone.cancelled() => break,
|
||||||
|
_ = tokio::time::sleep(randomized_cycle_delay()) => {}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
@@ -324,3 +330,29 @@ pub async fn store_data_usage_in_backend(
|
|||||||
attempts += 1;
|
attempts += 1;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use serial_test::serial;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
#[serial]
|
||||||
|
fn test_randomized_cycle_delay_keeps_configured_start_delay() {
|
||||||
|
// 120s with ±10% jitter should stay clearly above the historic 30s cap.
|
||||||
|
let delay = randomized_cycle_delay_for(Duration::from_secs(120));
|
||||||
|
assert!(delay > Duration::from_secs(30), "expected delay > 30s, got {delay:?}");
|
||||||
|
// Jitter window should stay within configured bounds.
|
||||||
|
assert!(delay >= Duration::from_secs(108));
|
||||||
|
assert!(delay <= Duration::from_secs(132));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
#[serial]
|
||||||
|
fn test_randomized_cycle_delay_handles_small_start_delay() {
|
||||||
|
// 0 is treated as minimum 1 second before jitter, with lower bound preserved.
|
||||||
|
let delay = randomized_cycle_delay_for(Duration::from_secs(0));
|
||||||
|
assert!(delay >= Duration::from_secs(1), "expected delay >= 1s");
|
||||||
|
assert!(delay < Duration::from_secs(2), "expected delay < 2s");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -137,6 +137,7 @@ impl SleepTimer {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use serial_test::serial;
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_scanner_speed_presets() {
|
fn test_scanner_speed_presets() {
|
||||||
@@ -166,6 +167,7 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test(start_paused = true)]
|
#[tokio::test(start_paused = true)]
|
||||||
|
#[serial]
|
||||||
async fn test_fastest_never_sleeps() {
|
async fn test_fastest_never_sleeps() {
|
||||||
let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed);
|
let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed);
|
||||||
SCANNER_IDLE_MODE.store(true, Ordering::Relaxed);
|
SCANNER_IDLE_MODE.store(true, Ordering::Relaxed);
|
||||||
@@ -179,6 +181,7 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test(start_paused = true)]
|
#[tokio::test(start_paused = true)]
|
||||||
|
#[serial]
|
||||||
async fn test_idle_mode_off_skips_sleep() {
|
async fn test_idle_mode_off_skips_sleep() {
|
||||||
let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed);
|
let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed);
|
||||||
SCANNER_IDLE_MODE.store(false, Ordering::Relaxed);
|
SCANNER_IDLE_MODE.store(false, Ordering::Relaxed);
|
||||||
|
|||||||
Reference in New Issue
Block a user