From 13f8768e7f7f15a3666fe802e153fbbeb354d525 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 10 Jul 2026 21:57:43 +0800 Subject: [PATCH] feat(ecstore): make replication timing intervals env-overridable for tests (#4680) Introduce RUSTFS_REPL_HEALTH_CHECK_INTERVAL_MS, RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS and RUSTFS_REPL_RESYNC_POLL_MAX_MS so tests and operators can shorten the replication background loops. Defaults are unchanged; invalid values fall back with a warn and values below 10ms are clamped to avoid busy-spin. Add replication_fast_env() e2e helper (backlog#1147 repl-4). --- crates/e2e_test/src/common.rs | 23 ++ .../ecstore/src/bucket/bucket_target_sys.rs | 5 +- crates/ecstore/src/bucket/replication/mod.rs | 1 + .../bucket/replication/replication_pool.rs | 25 +- .../bucket/replication/replication_timing.rs | 223 ++++++++++++++++++ 5 files changed, 268 insertions(+), 9 deletions(-) create mode 100644 crates/ecstore/src/bucket/replication/replication_timing.rs diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index fbeedabc4..1131ad2bd 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -514,6 +514,29 @@ impl Drop for RustFSTestEnvironment { } } +/// Short-interval replication timing overrides for fast e2e runs +/// (backlog#1147 repl-4). +/// +/// Returns the full set of replication background-loop env vars with +/// millisecond values so failure-recovery scenarios converge in seconds +/// instead of the production defaults (health check 5s, MRF flush 10s, +/// resync retry poll up to 60s). Pass to +/// [`RustFSTestEnvironment::start_rustfs_server_with_env`] as +/// `&replication_fast_env()`. The server reads each value once at background +/// task startup, so the vars must be set before the process starts. Values +/// below 10ms are clamped server-side to avoid busy-spin. +pub fn replication_fast_env() -> Vec<(&'static str, &'static str)> { + // Env var names mirror `crates/ecstore/src/bucket/replication/replication_timing.rs` + // (this is a process-boundary contract: the vars are passed to the spawned + // server, not consumed in-process). The names are pinned by the + // `env_var_names_are_a_cross_process_contract` test in that module. + vec![ + ("RUSTFS_REPL_HEALTH_CHECK_INTERVAL_MS", "200"), + ("RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS", "100"), + ("RUSTFS_REPL_RESYNC_POLL_MAX_MS", "250"), + ] +} + async fn execute_awscurl_with_service( url: &str, method: &str, diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 48376dacb..e7529977b 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -78,7 +78,6 @@ use tracing::warn; use url::Url; use uuid::Uuid; -const DEFAULT_HEALTH_CHECK_DURATION: Duration = Duration::from_secs(5); const DEFAULT_HEALTH_CHECK_RELOAD_DURATION: Duration = Duration::from_secs(30 * 60); const REDACTED_CREDENTIAL: &str = ""; @@ -335,7 +334,9 @@ impl BucketTargetSys { } pub async fn heartbeat(&self) { - let mut interval = tokio::time::interval(DEFAULT_HEALTH_CHECK_DURATION); + // Probe interval: `RUSTFS_REPL_HEALTH_CHECK_INTERVAL_MS` (default 5000ms, + // clamped to >=10ms), read once when the heartbeat task starts. + let mut interval = tokio::time::interval(crate::bucket::replication::replication_timing::health_check_interval()); loop { interval.tick().await; diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index c7723fec1..5e1aa0a28 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -39,6 +39,7 @@ mod replication_storage_boundary; mod replication_tagging_boundary; mod replication_target_boundary; mod replication_target_config_bridge; +pub(crate) mod replication_timing; mod replication_versioning_boundary; mod runtime_boundary; diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 48e5486bf..6107f50ae 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -738,7 +738,8 @@ impl ReplicationPool { /// Starts the MRF persister — ongoing background task. /// /// Drains `mrf_save_rx` (entries that overflowed the normal worker channels) and - /// writes them to the on-disk MRF file every 10 seconds or when 1 000 new + /// writes them to the on-disk MRF file every flush interval (default 10s, + /// overridable via `RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS`) or when 1 000 new /// entries accumulate. Each flush rewrites the whole cumulative backlog so no /// previously-persisted (and not-yet-replayed) entry is lost; the file is only /// consumed and cleared at startup. @@ -762,7 +763,9 @@ impl ReplicationPool { let mut flushed_len = 0usize; let mut dirty = false; let mut capped = false; - let mut interval = tokio::time::interval(Duration::from_secs(10)); + // Flush interval: `RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS` (default 10000ms, + // clamped to >=10ms), read once when the persister task starts. + let mut interval = tokio::time::interval(super::replication_timing::mrf_flush_interval()); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { @@ -992,6 +995,14 @@ impl ReplicationPool { /// Start the resync routine that runs in a loop async fn start_resync_routine(self: Arc, buckets: Vec, cancellation_token: CancellationToken) { + // Retry-poll sleep upper bound: `RUSTFS_REPL_RESYNC_POLL_MAX_MS` + // (default 60000ms, clamped to >=10ms), read once when this routine + // starts. The anti-busy-spin floor is min(1s, max) so the default + // keeps the historical "sleep at least one second" behavior while + // short test overrides stay short. + let max_sleep = super::replication_timing::resync_poll_max_sleep(); + let max_sleep_ms = u64::try_from(max_sleep.as_millis()).unwrap_or(u64::MAX).max(1); + let floor_sleep = Duration::from_secs(1).min(max_sleep); // Run the replication resync in a loop loop { let self_clone = self.clone(); @@ -1007,14 +1018,14 @@ impl ReplicationPool { } } - // Generate random duration between 0 and 1 minute + // Generate a random duration between 0 and `max_sleep` (default 1 minute) use rand::RngExt; - let duration_millis = rand::rng().random_range(0..60_000); + let duration_millis = rand::rng().random_range(0..max_sleep_ms); let mut duration = Duration::from_millis(duration_millis); - // Make sure to sleep at least a second to avoid high CPU ticks - if duration < Duration::from_secs(1) { - duration = Duration::from_secs(1); + // Make sure to sleep at least `floor_sleep` to avoid high CPU ticks + if duration < floor_sleep { + duration = floor_sleep; } tokio::time::sleep(duration).await; diff --git a/crates/ecstore/src/bucket/replication/replication_timing.rs b/crates/ecstore/src/bucket/replication/replication_timing.rs new file mode 100644 index 000000000..25f1468b8 --- /dev/null +++ b/crates/ecstore/src/bucket/replication/replication_timing.rs @@ -0,0 +1,223 @@ +// Copyright 2024 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. + +//! Env-overridable timing knobs for replication background loops +//! (backlog#1147 repl-4). +//! +//! Defaults are identical to the previously hardcoded values; the env vars +//! exist so tests and operators can shorten the loops (e.g. fast e2e runs). +//! Every value is read **once, when the owning background task starts** — +//! changing an env var mid-run does not affect already-running loops. +//! Invalid values fall back to the default with a `warn!`; values below +//! [`MIN_INTERVAL`] are clamped so an override of `0`/near-zero can never +//! busy-spin a loop (`tokio::time::interval` also panics on a zero period). + +use std::time::Duration; +use tracing::warn; + +use super::replication_logging::{LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION}; + +/// Overrides the remote-target health-check probe interval, in milliseconds. +/// Default: `5000` (5s). Test/ops tuning only. +pub(crate) const ENV_REPL_HEALTH_CHECK_INTERVAL_MS: &str = "RUSTFS_REPL_HEALTH_CHECK_INTERVAL_MS"; + +/// Overrides the MRF persistence flush interval, in milliseconds. +/// Default: `10000` (10s). Test/ops tuning only. +pub(crate) const ENV_REPL_MRF_FLUSH_INTERVAL_MS: &str = "RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS"; + +/// Overrides the upper bound of the random resync retry-poll sleep, in +/// milliseconds. Default: `60000` (60s). Test/ops tuning only. +pub(crate) const ENV_REPL_RESYNC_POLL_MAX_MS: &str = "RUSTFS_REPL_RESYNC_POLL_MAX_MS"; + +const DEFAULT_HEALTH_CHECK_INTERVAL: Duration = Duration::from_millis(5_000); +const DEFAULT_MRF_FLUSH_INTERVAL: Duration = Duration::from_millis(10_000); +const DEFAULT_RESYNC_POLL_MAX: Duration = Duration::from_millis(60_000); + +/// Hard floor for every replication timing override. Values below this are +/// clamped (with a `warn!`) so a pathological override cannot busy-spin a +/// background loop. The defaults are all far above this floor, so clamping +/// never alters default behavior. +pub(crate) const MIN_INTERVAL: Duration = Duration::from_millis(10); + +/// Remote-target health-check probe interval +/// ([`ENV_REPL_HEALTH_CHECK_INTERVAL_MS`], default 5s). +pub(crate) fn health_check_interval() -> Duration { + env_interval(ENV_REPL_HEALTH_CHECK_INTERVAL_MS, DEFAULT_HEALTH_CHECK_INTERVAL) +} + +/// MRF persistence flush interval +/// ([`ENV_REPL_MRF_FLUSH_INTERVAL_MS`], default 10s). +pub(crate) fn mrf_flush_interval() -> Duration { + env_interval(ENV_REPL_MRF_FLUSH_INTERVAL_MS, DEFAULT_MRF_FLUSH_INTERVAL) +} + +/// Upper bound of the random resync retry-poll sleep +/// ([`ENV_REPL_RESYNC_POLL_MAX_MS`], default 60s). +pub(crate) fn resync_poll_max_sleep() -> Duration { + env_interval(ENV_REPL_RESYNC_POLL_MAX_MS, DEFAULT_RESYNC_POLL_MAX) +} + +/// Resolve `name` as a millisecond interval: absent → `default` (silently); +/// unparsable → `default` with a `warn!`; parsed but below [`MIN_INTERVAL`] → +/// clamped to [`MIN_INTERVAL`] with a `warn!`. +fn env_interval(name: &str, default: Duration) -> Duration { + let raw = match std::env::var(name) { + Ok(raw) => raw, + Err(std::env::VarError::NotPresent) => return default, + Err(std::env::VarError::NotUnicode(_)) => { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + env = name, + default_ms = default.as_millis() as u64, + "replication timing override is not valid unicode — falling back to default" + ); + return default; + } + }; + match raw.trim().parse::() { + Ok(ms) => { + let value = Duration::from_millis(ms); + if value < MIN_INTERVAL { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + env = name, + value_ms = ms, + min_ms = MIN_INTERVAL.as_millis() as u64, + "replication timing override below minimum — clamping to avoid busy-spin" + ); + MIN_INTERVAL + } else { + value + } + } + Err(_) => { + warn!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + env = name, + value = %raw, + default_ms = default.as_millis() as u64, + "invalid replication timing override — falling back to default" + ); + default + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serial_test::serial; + use std::env; + + // SAFETY: this helper is only used from `#[serial]` tests, so no + // concurrent reader/writer can access the process environment while + // `env::set_var`/`env::remove_var` is active. + #[allow(unsafe_code)] + fn with_env(name: &str, value: Option<&str>, test_fn: F) { + let original = env::var_os(name); + match value { + Some(value) => unsafe { env::set_var(name, value) }, + None => unsafe { env::remove_var(name) }, + } + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(test_fn)); + match original { + Some(value) => unsafe { env::set_var(name, value) }, + None => unsafe { env::remove_var(name) }, + } + if let Err(e) = result { + std::panic::resume_unwind(e); + } + } + + /// The env var names are a cross-process contract: `crates/e2e_test/src/common.rs` + /// (`replication_fast_env`) passes them by literal name to spawned server + /// processes. Renaming one must fail here first. + #[test] + fn env_var_names_are_a_cross_process_contract() { + assert_eq!(ENV_REPL_HEALTH_CHECK_INTERVAL_MS, "RUSTFS_REPL_HEALTH_CHECK_INTERVAL_MS"); + assert_eq!(ENV_REPL_MRF_FLUSH_INTERVAL_MS, "RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS"); + assert_eq!(ENV_REPL_RESYNC_POLL_MAX_MS, "RUSTFS_REPL_RESYNC_POLL_MAX_MS"); + } + + #[test] + #[serial] + fn unset_env_uses_default() { + with_env(ENV_REPL_HEALTH_CHECK_INTERVAL_MS, None, || { + assert_eq!(health_check_interval(), Duration::from_secs(5)); + }); + with_env(ENV_REPL_MRF_FLUSH_INTERVAL_MS, None, || { + assert_eq!(mrf_flush_interval(), Duration::from_secs(10)); + }); + with_env(ENV_REPL_RESYNC_POLL_MAX_MS, None, || { + assert_eq!(resync_poll_max_sleep(), Duration::from_secs(60)); + }); + } + + #[test] + #[serial] + fn valid_override_is_applied() { + with_env(ENV_REPL_HEALTH_CHECK_INTERVAL_MS, Some("250"), || { + assert_eq!(health_check_interval(), Duration::from_millis(250)); + }); + with_env(ENV_REPL_MRF_FLUSH_INTERVAL_MS, Some("100"), || { + assert_eq!(mrf_flush_interval(), Duration::from_millis(100)); + }); + with_env(ENV_REPL_RESYNC_POLL_MAX_MS, Some("500"), || { + assert_eq!(resync_poll_max_sleep(), Duration::from_millis(500)); + }); + // Surrounding whitespace is tolerated. + with_env(ENV_REPL_MRF_FLUSH_INTERVAL_MS, Some(" 750 "), || { + assert_eq!(mrf_flush_interval(), Duration::from_millis(750)); + }); + } + + #[test] + #[serial] + fn invalid_override_falls_back_to_default() { + for bad in ["abc", "", "-5", "1.5", "10s", "99999999999999999999999999"] { + with_env(ENV_REPL_HEALTH_CHECK_INTERVAL_MS, Some(bad), || { + assert_eq!(health_check_interval(), Duration::from_secs(5), "input {bad:?}"); + }); + with_env(ENV_REPL_MRF_FLUSH_INTERVAL_MS, Some(bad), || { + assert_eq!(mrf_flush_interval(), Duration::from_secs(10), "input {bad:?}"); + }); + with_env(ENV_REPL_RESYNC_POLL_MAX_MS, Some(bad), || { + assert_eq!(resync_poll_max_sleep(), Duration::from_secs(60), "input {bad:?}"); + }); + } + } + + #[test] + #[serial] + fn zero_or_tiny_override_is_clamped_to_minimum() { + for tiny in ["0", "1", "9"] { + with_env(ENV_REPL_HEALTH_CHECK_INTERVAL_MS, Some(tiny), || { + assert_eq!(health_check_interval(), MIN_INTERVAL, "input {tiny:?}"); + }); + with_env(ENV_REPL_MRF_FLUSH_INTERVAL_MS, Some(tiny), || { + assert_eq!(mrf_flush_interval(), MIN_INTERVAL, "input {tiny:?}"); + }); + with_env(ENV_REPL_RESYNC_POLL_MAX_MS, Some(tiny), || { + assert_eq!(resync_poll_max_sleep(), MIN_INTERVAL, "input {tiny:?}"); + }); + } + // Exactly the minimum is accepted as-is. + with_env(ENV_REPL_HEALTH_CHECK_INTERVAL_MS, Some("10"), || { + assert_eq!(health_check_interval(), MIN_INTERVAL); + }); + } +}