diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b591772a..ca2187ecd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -86,6 +86,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Four-layer regression-prevention tests guard against silent feature deletion: compile-time module assertion, module-presence unit test, cross-module `Protocol` enum assertion, end-to-end SSH banner test against the running binary ### Changed +- **Scanner bucket walkers are no longer tied to the disk count**: a new scanner key `walkers_per_disk` (`RUSTFS_SCANNER_WALKERS_PER_DISK`, `1` through `16`, default `1`) spawns that many bucket walkers per local disk of an erasure set, so a single-disk deployment can walk several buckets at once instead of one bucket, one object at a time. The walkers of one disk share its scanner admission, so a peer still sees one scan session per physical disk, and remote disks keep one walker. The walker count is what `max_concurrent_disk_scans` caps and the disk count stays the floor; the `disk_concurrency_budget` log line now reports `walkers_per_disk`, `walkers`, and the `reason` for the effective limit. The default keeps today's I/O profile because a local disk cannot yet report whether it is rotational; the design intends a higher default on flash once it can (`docs/architecture/scanner-per-bucket-convergence.md`). As groundwork for pacing, the raw-enumeration page writer inserts into its building page in place and commits pages without copying them, and a regression test pins the cost of recording 20,000 entries in one prefix so the quadratic bookkeeping removed by [#8114](https://github.com/rustfs/rustfs/pull/8114) (rustfs#8113) cannot return. No persisted format, protocol or admin API changes beyond the additional config key, which older releases preserve and ignore. - **Encryption and KMS work merged since `1.0.0-rc.5`** (entries were missing from this section): - **Persisted KMS configuration secrets** are sealed field-by-field with `RUSTFS_KMS_CONFIG_SECRET`. **When the variable is unset the secrets are persisted in cleartext and the server only warns** (`persisted KMS configuration carries cleartext secrets`); it never refuses the write. Set it, identically, on every node, and re-save the configuration to seal an existing one. - **New v2 ciphertext frame format** with per-frame index binding and final-frame authentication. Its write switch `RUSTFS_ENCRYPTION_FRAME_V2` is **off by default**: v2 frames are unreadable by nodes without v2 read support, and encrypted ciphertext travels verbatim through transition, decommission and SSE-C replication passthrough, so turn it on only after every node — and every RustFS warm/replication target that receives raw ciphertext — runs a release with v2 read support. Reading v2 objects needs no switch. diff --git a/crates/config/README.md b/crates/config/README.md index 81600cff9..4cd84a44e 100644 --- a/crates/config/README.md +++ b/crates/config/README.md @@ -119,11 +119,19 @@ when restoring `/rustfs/console` for a custom OEM build. - `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` (canonical) - `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` (canonical) - `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` (canonical) +- `RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS` (canonical) +- `RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS` (canonical) +- `RUSTFS_SCANNER_WALKERS_PER_DISK` (canonical) Scanner cycle budget controls: - When `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS` is unset, the finite default is 1800 seconds (30 minutes), matching the scanner benchmark guidance. - An explicit `0` preserves the compatibility behavior of an unbounded runtime budget. Object and directory budgets likewise remain unbounded when explicitly set to `0`. +Scanner walker controls: + +- `RUSTFS_SCANNER_WALKERS_PER_DISK` sets how many bucket walkers each local disk of an erasure set runs; the default `1` walks one bucket per disk at a time, which is the right value for rotational media. On flash, `2` to `4` lets one disk walk several buckets concurrently, which matters most for single-disk deployments with many buckets, where the walk is otherwise fully serial. Values outside `1` through `16` are rejected by `admin config set`; an out-of-range or non-numeric environment value is logged as a scanner runtime config error (`state = "startup_apply_failed"`) and the scanner keeps the default. The walkers of one disk share that disk's scanner admission, so peers still see one scan session per physical disk, and `RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS` still caps the total. +- The same value is available as the persistent `scanner.walkers_per_disk` key and is reported under `runtime_config.walkers_per_disk` in the admin scanner status; see [Scanner Runtime Controls](../../docs/operations/scanner-runtime-controls.md). + - A timed-out cycle cancels cooperative scanner work, then fences its leader epoch before releasing the lease. An uncooperative I/O operation is dropped after the bounded shutdown window; its cursor is not claimed to be durable and the scanner reports `recovery-required` when the worker cannot stop cooperatively, the cycle state was not confirmed durable, or epoch fencing cannot be persisted. ## Mmap read environment aliases diff --git a/crates/config/src/constants/scanner.rs b/crates/config/src/constants/scanner.rs index 8c5b9231b..1465ad29c 100644 --- a/crates/config/src/constants/scanner.rs +++ b/crates/config/src/constants/scanner.rs @@ -59,6 +59,9 @@ pub const SCANNER_MAX_CONCURRENT_SET_SCANS: &str = "max_concurrent_set_scans"; /// Scanner config key capping concurrent scanner disk bucket walks per set. pub const SCANNER_MAX_CONCURRENT_DISK_SCANS: &str = "max_concurrent_disk_scans"; +/// Scanner config key setting how many bucket walkers each local disk runs. +pub const SCANNER_WALKERS_PER_DISK: &str = "walkers_per_disk"; + /// Scanner config key controlling how often object loops yield. pub const SCANNER_YIELD_EVERY_N_OBJECTS: &str = "yield_every_n_objects"; @@ -86,6 +89,7 @@ pub const SCANNER_KEYS: &[&str] = &[ SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_CONCURRENT_DISK_SCANS, + SCANNER_WALKERS_PER_DISK, SCANNER_YIELD_EVERY_N_OBJECTS, SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_ALERT_EXCESS_VERSION_SIZE, @@ -214,6 +218,14 @@ pub const ENV_SCANNER_MAX_CONCURRENT_SET_SCANS: &str = "RUSTFS_SCANNER_MAX_CONCU /// - Example: `export RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS=1` pub const ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS: &str = "RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS"; +/// Environment variable that sets how many bucket walkers each local disk runs. +/// The walkers of one disk share its scanner admission, so a peer still sees one +/// scan session per physical disk. Remote disks keep one walker. The product of +/// local disks and this factor is what `RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS` +/// caps; the disk count stays the floor. Valid range is `1` through `16`. +/// - Example: `export RUSTFS_SCANNER_WALKERS_PER_DISK=4` +pub const ENV_SCANNER_WALKERS_PER_DISK: &str = "RUSTFS_SCANNER_WALKERS_PER_DISK"; + /// Environment variable that controls how often scanner object loops yield to the async runtime. /// A value of `0` disables this extra object-count yield. /// - Example: `export RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS=32` @@ -228,6 +240,18 @@ pub const DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS: usize = 4; /// Default disk scan concurrency budget. pub const DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS: usize = 4; +/// Default bucket walkers per local disk. +/// +/// One walker per disk keeps today's I/O profile, which is what rotational +/// media needs. Flash is meant to default higher once a local disk can report +/// its media type; `rustfs_utils::os::Info::rotational` is never populated +/// today, so every deployment starts from one walker. +pub const DEFAULT_SCANNER_WALKERS_PER_DISK: usize = 1; + +/// Upper bound for `walkers_per_disk`; a larger factor only multiplies +/// contention on one spindle or NVMe queue without more disks to spread it. +pub const MAX_SCANNER_WALKERS_PER_DISK: usize = 16; + /// Default object interval for cooperative scanner yields. pub const DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS: u64 = 128; diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index b5dbc855a..f4235d514 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -2853,7 +2853,7 @@ mod tests { use rustfs_config::{ DEFAULT_DELIMITER, ENABLE_KEY, EnableState, HEAL_BITROT_CYCLE, HEAL_SUB_SYS, MYSQL_DSN_STRING, MYSQL_MAX_OPEN_CONNECTIONS, MYSQL_QUEUE_DIR, MYSQL_TABLE, SCANNER_CYCLE, SCANNER_IDLE_MODE, SCANNER_KEYS, SCANNER_SPEED, - SCANNER_SUB_SYS, + SCANNER_SUB_SYS, SCANNER_WALKERS_PER_DISK, }; #[tokio::test] @@ -4201,6 +4201,8 @@ mod tests { let value = match *key { SCANNER_SPEED => "fast", SCANNER_IDLE_MODE => "false", + // "1" is this key's default, which the diff encoding omits. + SCANNER_WALKERS_PER_DISK => "2", _ => "1", }; scanner_kvs.insert((*key).to_string(), value.to_string()); @@ -4222,6 +4224,8 @@ mod tests { let expected = match *key { SCANNER_SPEED => "fast", SCANNER_IDLE_MODE => "false", + // "1" is this key's default, which the diff encoding omits. + SCANNER_WALKERS_PER_DISK => "2", _ => "1", }; assert_eq!(decoded_scanner.get(key), expected, "scanner key {key} should survive roundtrip"); diff --git a/crates/ecstore/src/config/scanner.rs b/crates/ecstore/src/config/scanner.rs index 140f7c186..7d3f6e038 100644 --- a/crates/ecstore/src/config/scanner.rs +++ b/crates/ecstore/src/config/scanner.rs @@ -17,11 +17,11 @@ use rustfs_config::{ DEFAULT_SCANNER_BITROT_CYCLE_SECS, DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_IDLE_MODE, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, DEFAULT_SCANNER_SPEED, - DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, SCANNER_ALERT_EXCESS_FOLDERS, SCANNER_ALERT_EXCESS_VERSION_SIZE, - SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, - SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE, - SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED, SCANNER_START_DELAY, - SCANNER_YIELD_EVERY_N_OBJECTS, + DEFAULT_SCANNER_WALKERS_PER_DISK, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, SCANNER_ALERT_EXCESS_FOLDERS, + SCANNER_ALERT_EXCESS_VERSION_SIZE, SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, + SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, + SCANNER_IDLE_MODE, SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED, + SCANNER_START_DELAY, SCANNER_WALKERS_PER_DISK, SCANNER_YIELD_EVERY_N_OBJECTS, }; use std::sync::LazyLock; @@ -92,6 +92,11 @@ pub static DEFAULT_KVS: LazyLock = LazyLock::new(|| { value: DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS.to_string(), hidden_if_empty: false, }, + KV { + key: SCANNER_WALKERS_PER_DISK.to_owned(), + value: DEFAULT_SCANNER_WALKERS_PER_DISK.to_string(), + hidden_if_empty: false, + }, KV { key: SCANNER_YIELD_EVERY_N_OBJECTS.to_owned(), value: DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS.to_string(), diff --git a/crates/scanner/src/raw_page_index.rs b/crates/scanner/src/raw_page_index.rs index 023f9f585..dc8d75123 100644 --- a/crates/scanner/src/raw_page_index.rs +++ b/crates/scanner/src/raw_page_index.rs @@ -310,7 +310,8 @@ impl RawEnumerationPageIndex { return Err(RawEnumerationPageIndexError::StaleGeneration); } let entries_start = inner.validated_committed_entries()?.len(); - inner.commit_building_page(entries_start) + inner.commit_building_page(entries_start)?; + inner.pages.last().cloned().ok_or(RawEnumerationPageIndexError::EmptyCommit) } pub fn page(&self, page_index: u64, expected_digest: [u8; 32]) -> Result<&RawEnumerationPage, RawEnumerationPageIndexError> { @@ -347,8 +348,9 @@ impl RawEnumerationPageIndex { impl RawEnumerationPageIndexInner { // Callers validate the committed prefix at the restore/API boundary. The - // live writer maintains its entry count without revisiting that prefix. - fn commit_building_page(&mut self, entries_start: usize) -> Result { + // live writer maintains its entry count without revisiting that prefix, + // and the committed page moves into `pages` without a copy. + fn commit_building_page(&mut self, entries_start: usize) -> Result<(), RawEnumerationPageIndexError> { let Some(building) = self.building.as_ref() else { return Err(RawEnumerationPageIndexError::EmptyCommit); }; @@ -365,9 +367,9 @@ impl RawEnumerationPageIndexInner { }; let page = RawEnumerationPage::new(&self.parent, building); self.complete = page.terminal; - self.pages.push(page.clone()); + self.pages.push(page); self.generation = self.generation.saturating_add(1); - Ok(page) + Ok(()) } fn status(&self) -> RawEnumerationPageOwnerStatus { diff --git a/crates/scanner/src/raw_page_index/writer.rs b/crates/scanner/src/raw_page_index/writer.rs index 9cd7f04da..a6f93738d 100644 --- a/crates/scanner/src/raw_page_index/writer.rs +++ b/crates/scanner/src/raw_page_index/writer.rs @@ -112,8 +112,10 @@ impl RawEnumerationPageWriter { terminal: false, }); self.indexed.insert(entry.clone()); - building.entries.push(entry); - building.entries.sort(); + // The building page is normalized (strictly increasing) on every + // path that reaches here, so a positional insert keeps it so. + let position = building.entries.binary_search(&entry).unwrap_or_else(|position| position); + building.entries.insert(position, entry); building.terminal = false; self.inner.generation = self.inner.generation.saturating_add(1); } diff --git a/crates/scanner/src/remote_scanner.rs b/crates/scanner/src/remote_scanner.rs index 61036e1c8..932e73045 100644 --- a/crates/scanner/src/remote_scanner.rs +++ b/crates/scanner/src/remote_scanner.rs @@ -318,3 +318,60 @@ fn try_admit_remote_scanner_key(disk_key: String) -> Result, +} + +#[derive(Default)] +struct LocalDiskAdmissionState { + admission: Option, + holders: usize, +} + +/// One walker's stake in a [`LocalDiskAdmissionShare`]. +pub(crate) struct LocalDiskAdmissionHold { + share: Arc, +} + +impl LocalDiskAdmissionShare { + pub(crate) fn new() -> Arc { + Arc::new(Self { + state: Mutex::new(LocalDiskAdmissionState::default()), + }) + } + + pub(crate) fn acquire(self: &Arc, disk: &Disk) -> Result { + self.acquire_with(|| try_admit_remote_scanner(disk)) + } + + fn acquire_with( + self: &Arc, + admit: impl FnOnce() -> Result, + ) -> Result { + let mut state = self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + if state.admission.is_none() { + state.admission = Some(admit()?); + } + state.holders = state.holders.saturating_add(1); + Ok(LocalDiskAdmissionHold { share: Arc::clone(self) }) + } +} + +impl Drop for LocalDiskAdmissionHold { + fn drop(&mut self) { + let mut state = self.share.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + state.holders = state.holders.saturating_sub(1); + if state.holders == 0 { + state.admission = None; + } + } +} diff --git a/crates/scanner/src/remote_scanner/stream/tests.rs b/crates/scanner/src/remote_scanner/stream/tests.rs index 7b5e2df0b..06f2e6328 100644 --- a/crates/scanner/src/remote_scanner/stream/tests.rs +++ b/crates/scanner/src/remote_scanner/stream/tests.rs @@ -785,6 +785,50 @@ fn admission_rejects_busy_disk_before_replay_state_changes() { assert!(try_admit_remote_scanner_key(key).is_ok()); } +#[test] +fn local_disk_admission_share_holds_one_permit_until_the_last_walker_leaves() { + let key = format!("test-disk-{}", Uuid::new_v4()); + let share = LocalDiskAdmissionShare::new(); + let first = share + .acquire_with(|| try_admit_remote_scanner_key(key.clone())) + .expect("first walker acquires the disk permit"); + let second = share + .acquire_with(|| try_admit_remote_scanner_key(key.clone())) + .expect("second walker shares the held permit"); + assert!( + matches!(try_admit_remote_scanner_key(key.clone()), Err(ScannerError::RemoteDiskBusy)), + "a remote request sees one busy disk while any walker is active" + ); + drop(first); + assert!( + matches!(try_admit_remote_scanner_key(key.clone()), Err(ScannerError::RemoteDiskBusy)), + "the permit is released by the last holder, not the first" + ); + drop(second); + assert!(try_admit_remote_scanner_key(key).is_ok(), "no walker left: the disk is free again"); +} + +#[test] +fn local_disk_admission_share_retries_after_a_rejected_admission() { + let key = format!("test-disk-{}", Uuid::new_v4()); + let share = LocalDiskAdmissionShare::new(); + let remote = try_admit_remote_scanner_key(key.clone()).expect("remote request occupies the disk"); + assert!( + matches!( + share.acquire_with(|| try_admit_remote_scanner_key(key.clone())), + Err(ScannerError::RemoteDiskBusy) + ), + "walkers do not pre-empt an active remote session" + ); + drop(remote); + let hold = share + .acquire_with(|| try_admit_remote_scanner_key(key.clone())) + .expect("a rejected attempt leaves the share empty for a later retry"); + assert!(matches!(try_admit_remote_scanner_key(key.clone()), Err(ScannerError::RemoteDiskBusy))); + drop(hold); + assert!(try_admit_remote_scanner_key(key).is_ok()); +} + #[test] fn replay_cache_requires_contiguous_sequence_per_disk_session() { let session_id = Uuid::new_v4(); diff --git a/crates/scanner/src/runtime_config.rs b/crates/scanner/src/runtime_config.rs index cdace6a8f..6cf216092 100644 --- a/crates/scanner/src/runtime_config.rs +++ b/crates/scanner/src/runtime_config.rs @@ -20,17 +20,17 @@ use rustfs_config::{ DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE, DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS, DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_IDLE_MODE, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, - DEFAULT_SCANNER_SPEED, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, ENV_SCANNER_ALERT_EXCESS_FOLDERS, - ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, ENV_SCANNER_ALERT_EXCESS_VERSIONS, ENV_SCANNER_BITROT_CYCLE_SECS, - ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DIRECTORIES, + DEFAULT_SCANNER_SPEED, DEFAULT_SCANNER_WALKERS_PER_DISK, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, + ENV_SCANNER_ALERT_EXCESS_FOLDERS, ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, ENV_SCANNER_ALERT_EXCESS_VERSIONS, + ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DIRECTORIES, ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY, ENV_SCANNER_IDLE_MODE, ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED, - ENV_SCANNER_START_DELAY_SECS, ENV_SCANNER_YIELD_EVERY_N_OBJECTS, HEAL_BITROT_CYCLE, HEAL_SUB_SYS, - SCANNER_ALERT_EXCESS_FOLDERS, SCANNER_ALERT_EXCESS_VERSION_SIZE, SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, - SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, - SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE, SCANNER_MAX_CONCURRENT_DISK_SCANS, - SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED, SCANNER_START_DELAY, SCANNER_SUB_SYS, - SCANNER_YIELD_EVERY_N_OBJECTS, ScannerSpeed, + ENV_SCANNER_START_DELAY_SECS, ENV_SCANNER_WALKERS_PER_DISK, ENV_SCANNER_YIELD_EVERY_N_OBJECTS, HEAL_BITROT_CYCLE, + HEAL_SUB_SYS, MAX_SCANNER_WALKERS_PER_DISK, SCANNER_ALERT_EXCESS_FOLDERS, SCANNER_ALERT_EXCESS_VERSION_SIZE, + SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, + SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE, + SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED, SCANNER_START_DELAY, + SCANNER_SUB_SYS, SCANNER_WALKERS_PER_DISK, SCANNER_YIELD_EVERY_N_OBJECTS, ScannerSpeed, server_config::{Config as ServerConfig, KVS}, }; use serde::Serialize; @@ -53,6 +53,7 @@ const ENV_SCANNER_START_DELAY_SECS_DEPRECATED: &str = "RUSTFS_DATA_SCANNER_START const NO_DEFAULT_CYCLE_OVERRIDE: u64 = 0; const MAX_SCANNER_DELAY_FACTOR: f64 = 10_000.0; const SCANNER_DELAY_RANGE_REASON: &str = "expected scanner delay between 0 and 10000"; +const SCANNER_WALKERS_PER_DISK_RANGE_REASON: &str = "expected walkers per disk between 1 and 16"; static SCANNER_DEFAULT_CYCLE_SECS: AtomicU64 = AtomicU64::new(NO_DEFAULT_CYCLE_OVERRIDE); static SCANNER_RUNTIME_CONFIG_GENERATION: AtomicU64 = AtomicU64::new(0); @@ -96,6 +97,8 @@ pub(crate) struct ScannerRuntimeConfig { pub(crate) max_concurrent_set_scans_source: ScannerRuntimeConfigSource, pub(crate) max_concurrent_disk_scans: usize, pub(crate) max_concurrent_disk_scans_source: ScannerRuntimeConfigSource, + pub(crate) walkers_per_disk: usize, + pub(crate) walkers_per_disk_source: ScannerRuntimeConfigSource, pub(crate) yield_every_n_objects: u64, pub(crate) yield_every_n_objects_source: ScannerRuntimeConfigSource, pub(crate) alert_excess_versions: u64, @@ -138,6 +141,8 @@ impl Default for ScannerRuntimeConfig { max_concurrent_set_scans_source: ScannerRuntimeConfigSource::Default, max_concurrent_disk_scans: DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, max_concurrent_disk_scans_source: ScannerRuntimeConfigSource::Default, + walkers_per_disk: DEFAULT_SCANNER_WALKERS_PER_DISK, + walkers_per_disk_source: ScannerRuntimeConfigSource::Default, yield_every_n_objects: DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, yield_every_n_objects_source: ScannerRuntimeConfigSource::Default, alert_excess_versions: DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS, @@ -181,6 +186,7 @@ pub struct ScannerRuntimeConfigStatus { pub cache_save_timeout_seconds: ScannerRuntimeConfigValue, pub max_concurrent_set_scans: ScannerRuntimeConfigValue, pub max_concurrent_disk_scans: ScannerRuntimeConfigValue, + pub walkers_per_disk: ScannerRuntimeConfigValue, pub yield_every_n_objects: ScannerRuntimeConfigValue, pub alert_excess_versions: ScannerRuntimeConfigValue, pub alert_excess_version_size: ScannerRuntimeConfigValue, @@ -392,6 +398,9 @@ fn validate_persisted_scanner_runtime_config(config: &ServerConfig) -> Result<() validate_optional_config_u64(scanner_kvs, SCANNER_CACHE_SAVE_TIMEOUT, DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS)?; validate_optional_config_usize(scanner_kvs, SCANNER_MAX_CONCURRENT_SET_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS)?; validate_optional_config_usize(scanner_kvs, SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS)?; + if let Some(value) = config_value(scanner_kvs, SCANNER_WALKERS_PER_DISK, DEFAULT_SCANNER_WALKERS_PER_DISK) { + walkers_per_disk_from_count(SCANNER_WALKERS_PER_DISK, parse_config_usize(SCANNER_WALKERS_PER_DISK, value)?)?; + } validate_optional_config_u64(scanner_kvs, SCANNER_YIELD_EVERY_N_OBJECTS, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS)?; validate_optional_config_u64(scanner_kvs, SCANNER_ALERT_EXCESS_VERSIONS, DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS)?; validate_optional_config_u64(scanner_kvs, SCANNER_ALERT_EXCESS_VERSION_SIZE, DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE)?; @@ -484,6 +493,39 @@ fn cycle_duration_from_secs(key: &'static str, secs: u64) -> Result Result { + if (1..=MAX_SCANNER_WALKERS_PER_DISK).contains(&count) { + Ok(count) + } else { + Err(invalid_value(key, count.to_string(), SCANNER_WALKERS_PER_DISK_RANGE_REASON)) + } +} + +fn lookup_walkers_per_disk(kvs: Option<&KVS>) -> Result<(usize, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { + match rustfs_utils::get_env_parse_outcome::(ENV_SCANNER_WALKERS_PER_DISK) { + rustfs_utils::EnvParseOutcome::Parsed(count) => { + return walkers_per_disk_from_count(ENV_SCANNER_WALKERS_PER_DISK, count) + .map(|count| (count, ScannerRuntimeConfigSource::Env)); + } + rustfs_utils::EnvParseOutcome::Invalid => { + return Err(invalid_value( + ENV_SCANNER_WALKERS_PER_DISK, + "", + "expected unsigned integer count", + )); + } + rustfs_utils::EnvParseOutcome::Absent => {} + } + + if let Some(value) = config_value(kvs, SCANNER_WALKERS_PER_DISK, DEFAULT_SCANNER_WALKERS_PER_DISK) { + let count = parse_config_usize(SCANNER_WALKERS_PER_DISK, value)?; + return walkers_per_disk_from_count(SCANNER_WALKERS_PER_DISK, count) + .map(|count| (count, ScannerRuntimeConfigSource::Config)); + } + + Ok((DEFAULT_SCANNER_WALKERS_PER_DISK, ScannerRuntimeConfigSource::Default)) +} + fn lookup_start_delay(kvs: Option<&KVS>) -> Result<(Option, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { let aliases = [ENV_SCANNER_START_DELAY_SECS_DEPRECATED]; if let Some(secs) = rustfs_utils::get_env_opt_u64_with_aliases(ENV_SCANNER_START_DELAY_SECS, &aliases) { @@ -639,6 +681,7 @@ pub(crate) fn lookup_scanner_runtime_config( ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, )?; + let (walkers_per_disk, walkers_per_disk_source) = lookup_walkers_per_disk(scanner_kvs)?; let (yield_every_n_objects, yield_every_n_objects_source) = lookup_u64( scanner_kvs, SCANNER_YIELD_EVERY_N_OBJECTS, @@ -689,6 +732,8 @@ pub(crate) fn lookup_scanner_runtime_config( max_concurrent_set_scans_source, max_concurrent_disk_scans, max_concurrent_disk_scans_source, + walkers_per_disk, + walkers_per_disk_source, yield_every_n_objects, yield_every_n_objects_source, alert_excess_versions, @@ -825,6 +870,10 @@ pub fn scanner_runtime_config_status() -> ScannerRuntimeConfigStatus { value: config.max_concurrent_disk_scans, source: config.max_concurrent_disk_scans_source, }, + walkers_per_disk: ScannerRuntimeConfigValue { + value: config.walkers_per_disk, + source: config.walkers_per_disk_source, + }, yield_every_n_objects: ScannerRuntimeConfigValue { value: config.yield_every_n_objects, source: config.yield_every_n_objects_source, @@ -872,6 +921,10 @@ pub(crate) fn scanner_max_concurrent_disk_scans_configured() -> usize { current_scanner_runtime_config().max_concurrent_disk_scans } +pub(crate) fn scanner_walkers_per_disk_configured() -> usize { + current_scanner_runtime_config().walkers_per_disk +} + pub(crate) fn scanner_alert_excess_versions() -> u64 { current_scanner_runtime_config().alert_excess_versions } @@ -892,9 +945,10 @@ mod tests { use rustfs_config::{ DEFAULT_DELIMITER, DEFAULT_HEAL_BITROT_CYCLE_SECS, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY, - ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED, HEAL_BITROT_CYCLE, HEAL_SUB_SYS, SCANNER_BITROT_CYCLE, - SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, - SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE, SCANNER_SPEED, SCANNER_SUB_SYS, ScannerSpeed, + ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED, ENV_SCANNER_WALKERS_PER_DISK, HEAL_BITROT_CYCLE, HEAL_SUB_SYS, + SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, + SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE, SCANNER_SPEED, SCANNER_SUB_SYS, + SCANNER_WALKERS_PER_DISK, ScannerSpeed, }; use serial_test::serial; use std::collections::HashMap; @@ -1150,6 +1204,65 @@ mod tests { }); } + #[test] + #[serial] + fn scanner_runtime_config_walkers_per_disk_defaults_to_one_walker() { + let config = server_config_with_scanner(&[]); + + with_var_unset(ENV_SCANNER_WALKERS_PER_DISK, || { + let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config"); + + assert_eq!(resolved.walkers_per_disk, 1); + assert_eq!(resolved.walkers_per_disk_source, ScannerRuntimeConfigSource::Default); + }); + } + + #[test] + #[serial] + fn scanner_runtime_config_walkers_per_disk_accepts_persisted_range_and_rejects_zero_or_excess() { + with_var_unset(ENV_SCANNER_WALKERS_PER_DISK, || { + for (value, expected) in [("1", 1), ("4", 4), ("16", 16)] { + let config = server_config_with_scanner(&[(SCANNER_WALKERS_PER_DISK, value)]); + validate_scanner_runtime_config(&config).expect("walkers per disk inside the range should validate"); + let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config"); + assert_eq!(resolved.walkers_per_disk, expected, "persisted {value}"); + let expected_source = if expected == 1 { + ScannerRuntimeConfigSource::Default + } else { + ScannerRuntimeConfigSource::Config + }; + assert_eq!(resolved.walkers_per_disk_source, expected_source, "persisted {value}"); + } + for value in ["0", "17", "-1", "many"] { + let config = server_config_with_scanner(&[(SCANNER_WALKERS_PER_DISK, value)]); + let err = validate_scanner_runtime_config(&config).expect_err("walkers per disk outside the range must fail"); + assert!(err.to_string().contains("walkers_per_disk"), "persisted {value}: {err}"); + assert!( + lookup_scanner_runtime_config(Some(&config)).is_err(), + "persisted {value} must not resolve into a runtime config" + ); + } + }); + } + + #[test] + #[serial] + fn scanner_runtime_config_walkers_per_disk_env_overrides_and_is_range_checked() { + let config = server_config_with_scanner(&[(SCANNER_WALKERS_PER_DISK, "2")]); + + with_var(ENV_SCANNER_WALKERS_PER_DISK, Some("3"), || { + let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config"); + assert_eq!(resolved.walkers_per_disk, 3); + assert_eq!(resolved.walkers_per_disk_source, ScannerRuntimeConfigSource::Env); + }); + for value in ["0", "17", "x"] { + with_var(ENV_SCANNER_WALKERS_PER_DISK, Some(value), || { + let err = lookup_scanner_runtime_config(Some(&config)).expect_err("env walkers per disk is range checked"); + assert!(err.to_string().contains(ENV_SCANNER_WALKERS_PER_DISK), "env {value}: {err}"); + }); + } + } + #[test] fn scanner_runtime_config_validation_rejects_non_default_target() { let config = server_config_with_scanner_target("analytics", &[(SCANNER_SPEED, "slow")]); diff --git a/crates/scanner/src/scanner_folder/tests/incremental_enumeration.rs b/crates/scanner/src/scanner_folder/tests/incremental_enumeration.rs index 9390306db..56fc7be53 100644 --- a/crates/scanner/src/scanner_folder/tests/incremental_enumeration.rs +++ b/crates/scanner/src/scanner_folder/tests/incremental_enumeration.rs @@ -91,3 +91,42 @@ fn raw_enumeration_progress_invalid_entry_discards_index_but_retains_diagnostic_ assert!(progress.page_index().is_none()); assert_eq!(progress.cursor().expect("diagnostic cursor").entries_seen, 3); } + +// A flat prefix of tens of thousands of keys must cost the walker a bounded +// amount of bookkeeping per entry. The pre-incremental index re-normalized +// the cumulative entry list on every entry (rustfs#8113: 8,000 entries took +// about 40 s); the page writer hashes each entry once when its page commits. +#[test] +fn raw_enumeration_progress_record_entry_stays_linear_for_large_prefixes() { + const ENTRIES: usize = 20_000; + let committed_page_entries = ENTRIES - ENTRIES % SCANNER_RAW_ENUMERATION_PAGE_ENTRY_LIMIT; + let mut progress = RawEnumerationProgress::new("bucket", None); + RAW_PAGE_DIGEST_ENTRIES.set(0); + let start = Instant::now(); + // Reverse order is the worst case for the sorted insert into the building page. + for i in (0..ENTRIES).rev() { + progress.record_entry(&format!("entry-{i:06}")); + } + let elapsed = start.elapsed(); + assert_eq!(progress.checkpointable_entry_count(), ENTRIES); + assert_eq!( + RAW_PAGE_DIGEST_ENTRIES.get(), + committed_page_entries, + "each entry is hashed exactly once, when its page commits" + ); + assert!( + elapsed < Duration::from_secs(1), + "recording {ENTRIES} entries took {elapsed:?}; per-entry bookkeeping must stay amortised constant" + ); + let checkpoint = progress.page_index().expect("checkpoint with committed pages"); + assert_eq!( + RAW_PAGE_DIGEST_ENTRIES.get(), + ENTRIES, + "a checkpoint hashes only the building page it commits" + ); + assert_eq!(checkpoint.indexed_entries().expect("validated checkpoint").len(), ENTRIES); + assert_eq!( + progress.cursor().expect("cursor").entries_seen, + u64::try_from(ENTRIES).expect("entry count fits") + ); +} diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 7c2e169cc..fb1c0659a 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -26,7 +26,7 @@ use futures::future::join_all; use metrics::counter; use rand::seq::SliceRandom as _; #[cfg(test)] -use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS}; +use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_WALKERS_PER_DISK}; use rustfs_data_usage::{BucketTargetUsageInfo, BucketUsageInfo, observed_data_usage_is_newer}; use rustfs_filemeta::FileMeta; use rustfs_heal_contracts::heal_channel::HealScanMode; @@ -1182,6 +1182,42 @@ fn namespace_scanner_workers( workers } +struct NamespaceScannerWalker { + disk: T, + mode: NamespaceScannerWorkerMode, + /// Shared by every walker of one local disk; `None` for disks the + /// coordinator reaches over the network, which keep today's one walker. + local_admission: Option>, +} + +/// Expands each worker into its bucket walkers. A local coordinator disk gets +/// `walkers_per_disk` walkers sharing one disk admission; every other worker +/// keeps a single walker, so the walker count never drops below the disk count. +fn namespace_scanner_walkers( + workers: Vec<(T, NamespaceScannerWorkerMode)>, + walkers_per_disk: usize, + is_local: impl Fn(&T) -> bool, +) -> Vec> { + let mut walkers = Vec::with_capacity(workers.len()); + for (disk, mode) in workers { + if mode == NamespaceScannerWorkerMode::Coordinator && is_local(&disk) { + let local_admission = crate::remote_scanner::LocalDiskAdmissionShare::new(); + walkers.extend((0..walkers_per_disk.max(1)).map(|_| NamespaceScannerWalker { + disk: disk.clone(), + mode, + local_admission: Some(local_admission.clone()), + })); + } else { + walkers.push(NamespaceScannerWalker { + disk, + mode, + local_admission: None, + }); + } + } + walkers +} + fn group_remote_disks_by_peer(disks: Vec, peer_key: impl Fn(&T) -> String) -> Vec> { let mut groups = HashMap::>::new(); for disk in disks { diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index d9915153d..d09fae54a 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -470,7 +470,11 @@ impl ScannerIOCache for SetDisks { "Scanner set assigned remote disks without namespace scanner support to coordinator-driven workers" ); } - let dispatch_limit = scanner_max_concurrent_disk_scans(workers.len()); + let online_disks = workers.len(); + let walkers_per_disk = crate::runtime_config::scanner_walkers_per_disk_configured(); + let walkers = namespace_scanner_walkers(workers, walkers_per_disk, |disk| disk.is_local()); + let dispatch_limit = scanner_max_concurrent_disk_scans(walkers.len()); + let configured_disk_scans = crate::runtime_config::scanner_max_concurrent_disk_scans_configured(); let serial_progress_accounting = budget.requires_serial_progress_accounting(); let disk_scan_limit = scanner_budgeted_concurrency_limit(dispatch_limit, serial_progress_accounting); record_disk_scan_concurrency_limit(&pool_label, &set_label, disk_scan_limit); @@ -480,10 +484,18 @@ impl ScannerIOCache for SetDisks { component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_IO, state = "disk_concurrency_budget", - reason = if serial_progress_accounting { "count_budget" } else { "configured_or_available_work" }, + reason = if serial_progress_accounting { + "count_budget" + } else if configured_disk_scans != 0 && configured_disk_scans < walkers.len() { + "configured_cap" + } else { + "available_walkers" + }, pool = self.pool_index, set = self.set_index, - online_disks = workers.len(), + online_disks, + walkers_per_disk, + walkers = walkers.len(), dispatch_limit, concurrency_limit = disk_scan_limit, "Scanner disk concurrency budget resolved" @@ -629,7 +641,7 @@ impl ScannerIOCache for SetDisks { let cache_mutex: Arc> = Arc::new(Mutex::new(cache)); - let (bucket_result_tx, mut bucket_result_rx) = mpsc::channel::(workers.len()); + let (bucket_result_tx, mut bucket_result_rx) = mpsc::channel::(walkers.len()); let cache_mutex_clone = cache_mutex.clone(); let ctx_clone = ctx.clone(); @@ -662,7 +674,12 @@ impl ScannerIOCache for SetDisks { let bucket_rx_mutex: Arc>> = Arc::new(Mutex::new(bucket_rx)); let remaining_bucket_work = Arc::new(AtomicUsize::new(buckets.len())); let bucket_work_complete = CancellationToken::new(); - for (disk, worker_mode) in workers { + for NamespaceScannerWalker { + disk, + mode: worker_mode, + local_admission, + } in walkers + { let service_cohort_clone = service_cohort.clone(); let bucket_rx_mutex_clone = bucket_rx_mutex.clone(); let bucket_tx_clone = bucket_tx.clone(); @@ -907,8 +924,8 @@ impl ScannerIOCache for SetDisks { continue; } - let _local_admission = if disk.is_local() { - match crate::remote_scanner::try_admit_remote_scanner(&disk) { + let _local_admission = if let Some(share) = &local_admission { + match share.acquire(&disk) { Ok(admission) => Some(admission), Err(e) => { if requeue_bucket_work(&bucket_tx_clone, &bucket, &mut work_guard).await { @@ -936,6 +953,8 @@ impl ScannerIOCache for SetDisks { } else { None }; + #[cfg(test)] + super::tests::observe_coordinator_bucket_scan_admitted(&bucket.name).await; // Lock order: scanner leader fence -> set-scoped per-bucket cache lock -> // cache object read/write. diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index dfcd4fc4d..c2ba1a13d 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -4651,3 +4651,173 @@ async fn replication_maintenance_refreshes_compacted_stats_across_cycles_and_ret clear_dirty_usage_buckets_for_tests(); } } + +pub(super) struct CoordinatorBucketScanProbe { + active: AtomicUsize, + peak: AtomicUsize, + hold: Duration, +} + +static COORDINATOR_BUCKET_SCAN_PROBE: std::sync::Mutex>> = std::sync::Mutex::new(None); + +/// Called by the coordinator walker loop once a bucket scan is admitted on +/// its disk. The hook is absent from non-test builds and a no-op until a test +/// installs a probe; the probe then holds the walker so overlapping +/// admissions become observable as a peak. +pub(super) async fn observe_coordinator_bucket_scan_admitted(_bucket: &str) { + let probe = COORDINATOR_BUCKET_SCAN_PROBE + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone(); + let Some(probe) = probe else { return }; + let active = probe.active.fetch_add(1, Ordering::SeqCst) + 1; + probe.peak.fetch_max(active, Ordering::SeqCst); + tokio::time::sleep(probe.hold).await; + probe.active.fetch_sub(1, Ordering::SeqCst); +} + +fn install_coordinator_bucket_scan_probe(hold: Duration) -> Arc { + let probe = Arc::new(CoordinatorBucketScanProbe { + active: AtomicUsize::new(0), + peak: AtomicUsize::new(0), + hold, + }); + *COORDINATOR_BUCKET_SCAN_PROBE + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(probe.clone()); + probe +} + +fn clear_coordinator_bucket_scan_probe() { + *COORDINATOR_BUCKET_SCAN_PROBE + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = None; +} + +#[test] +fn namespace_scanner_walkers_multiply_local_disks_and_share_their_admission() { + let server_epoch = Uuid::from_u128(7); + let workers = vec![ + ("local-a", NamespaceScannerWorkerMode::Coordinator), + ("local-b", NamespaceScannerWorkerMode::Coordinator), + ("legacy-peer", NamespaceScannerWorkerMode::Coordinator), + ("v4-peer", NamespaceScannerWorkerMode::RemoteV4(server_epoch)), + ]; + let is_local = |disk: &&str| disk.starts_with("local"); + + let walkers = namespace_scanner_walkers(workers.clone(), 3, is_local); + assert_eq!(walkers.len(), 2 * 3 + 2, "two local disks fan out, both remote disks keep one walker"); + let local_a: Vec<_> = walkers.iter().filter(|walker| walker.disk == "local-a").collect(); + assert_eq!(local_a.len(), 3); + let share = local_a[0] + .local_admission + .as_ref() + .expect("local walkers carry an admission share"); + assert!( + local_a + .iter() + .all(|walker| walker.local_admission.as_ref().is_some_and(|other| Arc::ptr_eq(other, share))), + "walkers of one disk share one admission" + ); + let local_b = walkers + .iter() + .find(|walker| walker.disk == "local-b") + .expect("local-b walker"); + assert!( + !Arc::ptr_eq(local_b.local_admission.as_ref().expect("local-b share"), share), + "each disk owns its own admission share" + ); + assert!( + walkers + .iter() + .filter(|walker| !is_local(&walker.disk)) + .all(|walker| walker.local_admission.is_none()), + "remote disks admit through the peer, not a local share" + ); + assert_eq!( + walkers + .iter() + .filter(|walker| walker.mode == NamespaceScannerWorkerMode::RemoteV4(server_epoch)) + .count(), + 1 + ); + // The walker count, not the disk count, is the available work the + // configured cap applies to; the disk count stays the floor. + assert_eq!(scanner_concurrency_limit(0, walkers.len()), 8); + assert_eq!(scanner_concurrency_limit(4, walkers.len()), 4); + for factor in [0, 1] { + let walkers = namespace_scanner_walkers(workers.clone(), factor, is_local); + assert_eq!(walkers.len(), workers.len(), "factor {factor} keeps one walker per disk"); + assert_eq!(scanner_concurrency_limit(0, walkers.len()), workers.len()); + } +} + +async fn run_full_cycle(store: &Arc, cycle: u64) -> ScannerCycleStatus { + let ctx = CancellationToken::new(); + let budget = ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()); + let (updates, _receiver) = mpsc::channel(8); + tokio::time::timeout( + Duration::from_secs(60), + nsscanner_with_storage_status_scoped( + store.as_ref(), + ScannerCycleRequest { + ctx, + budget, + updates, + want_cycle: cycle, + leader_epoch: 11, + scan_mode: HealScanMode::Normal, + scan_scope: ScannerBucketScanScope::default(), + persisted_usage_baseline: None, + observed_usage_candidate: None, + requires_full_scan: true, + resolved_scope_observer: None, + service_cohort: None, + }, + ), + ) + .await + .expect("cycle should finish within the test deadline") + .expect("cycle should succeed") + .status +} + +#[tokio::test] +#[serial] +async fn walkers_per_disk_lets_one_disk_walk_two_buckets_concurrently() { + let (_temp_dir, store) = setup_local_scanner_store(1, 1).await; + clear_dirty_usage_buckets_for_tests(); + for bucket in ["walk-a", "walk-b"] { + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + let mut reader = ScannerPutObjReader::from_vec(b"object".to_vec()); + store + .put_object(bucket, "object", &mut reader, &ScannerObjectOptions::default()) + .await + .expect("object should persist"); + } + wait_for_namespace_commit_tails(store.as_ref()).await; + + // The probe holds every admitted bucket scan long enough that two walkers + // on the one disk must overlap; a single walker cannot overlap with itself. + let hold = Duration::from_millis(750); + for (cycle, factor, expected_peak) in [(1, "1", 1), (2, "2", 2)] { + let probe = install_coordinator_bucket_scan_probe(hold); + let status = temp_env::async_with_vars([(ENV_SCANNER_WALKERS_PER_DISK, Some(factor))], async { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); + run_full_cycle(&store, cycle).await + }) + .await; + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); + clear_coordinator_bucket_scan_probe(); + assert_eq!(status, ScannerCycleStatus::Complete, "walkers_per_disk={factor}"); + assert_eq!( + probe.peak.load(Ordering::SeqCst), + expected_peak, + "walkers_per_disk={factor}: concurrent bucket scans admitted on the single disk" + ); + assert_eq!(probe.active.load(Ordering::SeqCst), 0, "every admitted scan left the probe"); + } +} diff --git a/docs/architecture/scanner-per-bucket-convergence.md b/docs/architecture/scanner-per-bucket-convergence.md index 416cfd788..063fe043d 100644 --- a/docs/architecture/scanner-per-bucket-convergence.md +++ b/docs/architecture/scanner-per-bucket-convergence.md @@ -253,9 +253,9 @@ for within-cycle reuse. ## Pacing and worker concurrency -Today the number of bucket walkers per erasure set equals the number of -disks in the set, capped by `max_concurrent_disk_scans`, and -`scanner_concurrency_limit` takes the minimum of the configured value and the +Before phase 3 the number of bucket walkers per erasure set equalled the +number of disks in the set, capped by `max_concurrent_disk_scans`, and +`scanner_concurrency_limit` took the minimum of the configured value and the disk count. Buckets queue on one channel per set and each walker takes one bucket at a time; inside a bucket the folder walk is sequential and `DynamicSleeper` sleeps after every object for the work time multiplied by @@ -273,7 +273,14 @@ resource ceiling. The scanner adopts the same shape: count multiplied by a per-disk walker factor. The factor defaults from disk type: 1 for rotational media, higher for flash, where xl.meta decode is CPU bound and there is no seek penalty. Disk count stays the floor so - distributed deployments keep at least today's parallelism. + distributed deployments keep at least today's parallelism. Shipped as + `scanner.walkers_per_disk` (`RUSTFS_SCANNER_WALKERS_PER_DISK`, `1` through + `16`): each local coordinator disk gets that many walkers, which share one + per-disk scanner admission so a peer still sees one scan session per + physical disk; remote disks keep one walker. The default is `1` everywhere + until a local disk can report its media type + (`rustfs_utils::os::Info::rotational` is never populated today); the flash + default above `1` remains the intent. - **Throttle**: the `DynamicSleeper` factor is the pacer's output, not a fixed preset. `ScannerSpeed` presets remain as operator overrides and as the initial value. Foreground workload backoff in `sleeper.rs` remains the @@ -298,7 +305,7 @@ not tuning advice. | Pacer raises walkers and lowers sleep | CPU and I/O ceilings are absolute. Foreground admission backoff cannot be undercut. Walker count never exceeds the ceiling-derived maximum even when the target is missed. | | Faster convergence discovers heal and replication work sooner | Heal admission keeps its queue bounds and `queue_full` and `dropped` accounting in `crates/scanner/src/scanner_folder/ledger.rs`; replication MRF keeps its pending cap. Convergence speed does not change retry intervals. Discovered work is swept lazily under those bounds, in the sense of a lazy sweeper: discovery and execution are decoupled. | | Guardrail vetoes publication | A veto is attributed to a field and localised to a peer where the field allows it; only movement and topology defer the whole cycle. Deferred retries use the capped superseded backoff. A probe failure on one peer never costs more than that peer's dirtied buckets. | -| One bucket walk may be large | Inside a bucket, the walk takes a per-prefix quantum: a prefix that exhausts it checkpoints and yields to the next prefix or bucket, and resumes under the existing cursor and page-index fences. No single prefix may consume the whole cycle budget. The walker's own bookkeeping must stay linear in entries per prefix; `RawEnumerationProgress::record_entry` in `crates/scanner/src/scanner_folder.rs` copies the accumulated entry list on every entry today and is a prerequisite fix for pacing. | +| One bucket walk may be large | Inside a bucket, the walk takes a per-prefix quantum: a prefix that exhausts it checkpoints and yields to the next prefix or bucket, and resumes under the existing cursor and page-index fences. No single prefix may consume the whole cycle budget. The walker's own bookkeeping must stay linear in entries per prefix. `RawEnumerationProgress::record_entry` in `crates/scanner/src/scanner_folder.rs` once re-normalized the accumulated entry list on every entry (rustfs#8113: 8,000 entries cost about 40 s); since #8114 the page writer hashes each entry once when its page commits, and a regression test pins 20,000 entries in one prefix to that cost so pacing cannot be built on a quadratic walker. | ## What stays fail-closed @@ -444,10 +451,12 @@ modes. a hundred thousand entries no longer starves the buckets after it; a multi-node deployment with one peer's probe failing publishes every bucket that peer did not dirty. -3. **Pacer and walker concurrency.** Fix the quadratic entry bookkeeping in - `RawEnumerationProgress::record_entry` first, then decouple walker count - from disk count and drive the throttle from the pacing target under CPU - and I/O ceilings. Oracle: on a single flash disk the same namespace +3. **Pacer and walker concurrency.** Pin the entry bookkeeping in + `RawEnumerationProgress::record_entry` to linear cost first (the + clone-per-entry itself was removed by #8114; the regression test and the + in-place page writer are this phase's groundwork), then decouple walker + count from disk count (`walkers_per_disk`, shipped) and drive the throttle + from the pacing target under CPU and I/O ceilings (open). Oracle: on a single flash disk the same namespace converges in a fraction of today's cycle time with scanner CPU under the ceiling; on rotational media the default factor keeps today's I/O profile; a foreground write load at the admission threshold still backs diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index 2029bc7d7..65a98abce 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -43,7 +43,8 @@ Bitrot cycle resolution differs because the canonical persistent key belongs to | `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.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.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. | -| `scanner.max_concurrent_disk_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS`) | Caps concurrent disk bucket walks per set. `0` keeps disk-count-derived concurrency. | +| `scanner.max_concurrent_disk_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS`) | Caps concurrent disk bucket walks per set. `0` keeps walker-count-derived concurrency. | +| `scanner.walkers_per_disk` | `RUSTFS_SCANNER_WALKERS_PER_DISK` | count | `1` (`DEFAULT_SCANNER_WALKERS_PER_DISK`) | Bucket walkers spawned per local disk of a set. The walkers of one disk share its scanner admission, so a peer still sees one scan session per physical disk; remote disks keep one walker. The resulting walker count is what `scanner.max_concurrent_disk_scans` caps, and the disk count stays the floor. Valid range is `1` through `16` (`MAX_SCANNER_WALKERS_PER_DISK`); `1` keeps today's I/O profile and is the right value for rotational media. The `disk_concurrency_budget` log line reports `walkers_per_disk`, `walkers`, the effective `concurrency_limit`, and its `reason` (`available_walkers`, `configured_cap`, or `count_budget`). | | `scanner.yield_every_n_objects` | `RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS` | objects | `128` (`DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS`) | Controls how often object loops yield to the async runtime. `0` disables this extra yield. | | `scanner.alert_excess_versions` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS` | versions | `100` (`DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS`) | Version count threshold for scanner alerts. | | `scanner.alert_excess_version_size` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE` | bytes | `1099511627776` (`DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE`) | Retained version byte threshold for scanner alerts. | @@ -61,6 +62,16 @@ Speed presets (`crates/config/src/constants/scanner.rs`) set the base sleep mult 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. +### Walkers per disk + +The scanner walks the buckets of an erasure set with one walker per local disk by default, so a single-disk deployment walks one bucket at a time no matter how `scanner.max_concurrent_disk_scans` is set. `scanner.walkers_per_disk` (`RUSTFS_SCANNER_WALKERS_PER_DISK`) multiplies each local disk into that many walkers. + +- Keep the default `1` on rotational media: several walkers on one spindle compete for seeks and slow every walk down. +- On flash, `2` to `4` is the useful range. Raise it when the scanner status shows cycles ending on the cycle budget (`metrics.pacing_pressure.last_cycle_budget_limited`) or when many buckets are never reached within a cycle, and the disk has idle I/O capacity. +- The walkers of one disk share that disk's scanner admission, so a peer still sees one scan session per physical disk; remote disks keep one walker. `scanner.max_concurrent_disk_scans` still caps the resulting walker count, and the disk count stays the floor. +- Confirm the setting took effect from `runtime_config.walkers_per_disk.value` in the admin scanner status and from the `disk_concurrency_budget` log line, which reports `walkers_per_disk`, `walkers`, the effective `concurrency_limit`, and the `reason` it was chosen. +- Helm: `config.rustfs.scanner.walkers_per_disk`; Docker or systemd: the environment variable. The key is additive, so an older binary ignores it. + ### Environment-only scanner knobs These have no persistent key and are read from the environment only. @@ -144,6 +155,7 @@ runtime_config.delay.value runtime_config.max_wait_seconds.value runtime_config.cycle_interval_seconds.value runtime_config.bitrot_cycle_seconds.value +runtime_config.walkers_per_disk.value cycle_schedule.effective_interval_seconds cycle_schedule.clean_idle_backoff_enabled cycle_schedule.clean_idle_backoff_multiplier diff --git a/helm/README.md b/helm/README.md index b4cec75c0..42d5b3476 100644 --- a/helm/README.md +++ b/helm/README.md @@ -140,6 +140,7 @@ behavior. | config.rustfs.scanner.cache_save_timeout_secs | string | `""` | Override scanner cache save timeout in seconds with `RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS` (minimum `1`). | | config.rustfs.scanner.max_concurrent_set_scans | string | `""` | Cap concurrent scanner set tasks with `RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS` (`0` keeps topology-derived concurrency). | | config.rustfs.scanner.max_concurrent_disk_scans | string | `""` | Cap concurrent scanner disk bucket walks per set with `RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS` (`0` keeps disk-count-derived concurrency). | +| config.rustfs.scanner.walkers_per_disk | string | `""` | Bucket walkers per local disk with `RUSTFS_SCANNER_WALKERS_PER_DISK` (`1` to `16`; the default `1` keeps one bucket walk per disk, raise it on flash so one disk walks several buckets concurrently). | | config.rustfs.scanner.yield_every_n_objects | string | `""` | Yield to the async runtime every N scanned objects with `RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS` (`0` disables extra yield). | | config.rustfs.scanner.alert_excess_versions | string | `""` | Set version count threshold for scanner alerts with `RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS`. | | config.rustfs.scanner.alert_excess_version_size | string | `""` | Set retained version byte threshold for scanner alerts with `RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE`. | diff --git a/helm/rustfs/templates/configmap.yaml b/helm/rustfs/templates/configmap.yaml index e5aaa352f..f22e296c9 100644 --- a/helm/rustfs/templates/configmap.yaml +++ b/helm/rustfs/templates/configmap.yaml @@ -123,6 +123,9 @@ data: {{- end }} {{- if ne (toString .max_concurrent_disk_scans) "" }} RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS: {{ .max_concurrent_disk_scans | quote }} + {{- end }} + {{- if ne (toString .walkers_per_disk) "" }} + RUSTFS_SCANNER_WALKERS_PER_DISK: {{ .walkers_per_disk | quote }} {{- end }} {{- if ne (toString .yield_every_n_objects) "" }} RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS: {{ .yield_every_n_objects | quote }} diff --git a/helm/rustfs/values.yaml b/helm/rustfs/values.yaml index de0c4b7f1..b7b622d7e 100644 --- a/helm/rustfs/values.yaml +++ b/helm/rustfs/values.yaml @@ -185,6 +185,8 @@ config: max_concurrent_set_scans: "" # Cap concurrent scanner disk bucket walks per set (0 keeps disk-count-derived concurrency) max_concurrent_disk_scans: "" + # Bucket walkers per local disk (1-16; default 1 keeps one bucket walk per disk, raise on flash) + walkers_per_disk: "" # Yield to the async runtime every N scanned objects (0 disables extra yield) yield_every_n_objects: "" # Version count threshold for scanner alerts diff --git a/rustfs/src/admin/handlers/config_admin.rs b/rustfs/src/admin/handlers/config_admin.rs index 15bcd7763..3344d4643 100644 --- a/rustfs/src/admin/handlers/config_admin.rs +++ b/rustfs/src/admin/handlers/config_admin.rs @@ -68,15 +68,15 @@ use rustfs_config::{ ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DIRECTORIES, ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY, ENV_SCANNER_IDLE_MODE, ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED, - ENV_SCANNER_START_DELAY_SECS, ENV_SCANNER_YIELD_EVERY_N_OBJECTS, HEAL_BITROT_CYCLE, HEAL_SUB_SYS, - MAX_ADMIN_REQUEST_BODY_SIZE, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, + ENV_SCANNER_START_DELAY_SECS, ENV_SCANNER_WALKERS_PER_DISK, ENV_SCANNER_YIELD_EVERY_N_OBJECTS, HEAL_BITROT_CYCLE, + HEAL_SUB_SYS, MAX_ADMIN_REQUEST_BODY_SIZE, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, SCANNER_ALERT_EXCESS_FOLDERS, SCANNER_ALERT_EXCESS_VERSION_SIZE, SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE, SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED, - SCANNER_START_DELAY, SCANNER_SUB_SYS, SCANNER_YIELD_EVERY_N_OBJECTS, WEBHOOK_AUTH_TOKEN, WEBHOOK_BATCH_SIZE, - WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_HTTP_TIMEOUT, WEBHOOK_MAX_RETRY, WEBHOOK_QUEUE_DIR, - WEBHOOK_QUEUE_LIMIT, WEBHOOK_RETRY_INTERVAL, + SCANNER_START_DELAY, SCANNER_SUB_SYS, SCANNER_WALKERS_PER_DISK, SCANNER_YIELD_EVERY_N_OBJECTS, WEBHOOK_AUTH_TOKEN, + WEBHOOK_BATCH_SIZE, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_HTTP_TIMEOUT, WEBHOOK_MAX_RETRY, + WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_RETRY_INTERVAL, }; use rustfs_credentials::Credentials; use rustfs_policy::policy::action::{Action, AdminAction}; @@ -286,6 +286,12 @@ const SCANNER_HELP_KEYS: &[HelpKeyMetadata] = &[ description: "cap concurrent disk bucket walks per set, 0 uses disk-count defaults", optional: true, }, + HelpKeyMetadata { + key: SCANNER_WALKERS_PER_DISK, + type_name: "number", + description: "bucket walkers per local disk, 1 through 16; remote disks keep one walker", + optional: true, + }, HelpKeyMetadata { key: SCANNER_YIELD_EVERY_N_OBJECTS, type_name: "number", @@ -1619,6 +1625,7 @@ fn env_help_key(sub_system: &str, key: &str) -> String { (SCANNER_SUB_SYS, SCANNER_CACHE_SAVE_TIMEOUT) => ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS.to_string(), (SCANNER_SUB_SYS, SCANNER_MAX_CONCURRENT_SET_SCANS) => ENV_SCANNER_MAX_CONCURRENT_SET_SCANS.to_string(), (SCANNER_SUB_SYS, SCANNER_MAX_CONCURRENT_DISK_SCANS) => ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS.to_string(), + (SCANNER_SUB_SYS, SCANNER_WALKERS_PER_DISK) => ENV_SCANNER_WALKERS_PER_DISK.to_string(), (SCANNER_SUB_SYS, SCANNER_YIELD_EVERY_N_OBJECTS) => ENV_SCANNER_YIELD_EVERY_N_OBJECTS.to_string(), (SCANNER_SUB_SYS, SCANNER_ALERT_EXCESS_VERSIONS) => ENV_SCANNER_ALERT_EXCESS_VERSIONS.to_string(), (SCANNER_SUB_SYS, SCANNER_ALERT_EXCESS_VERSION_SIZE) => ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE.to_string(), diff --git a/scripts/test_helm_templates.sh b/scripts/test_helm_templates.sh index f273430b9..e6e7907d0 100755 --- a/scripts/test_helm_templates.sh +++ b/scripts/test_helm_templates.sh @@ -180,6 +180,7 @@ scanner_config_output=$(render_chart \ --set config.rustfs.scanner.cache_save_timeout_secs=30 \ --set config.rustfs.scanner.max_concurrent_set_scans=2 \ --set config.rustfs.scanner.max_concurrent_disk_scans=1 \ + --set config.rustfs.scanner.walkers_per_disk=2 \ --set config.rustfs.scanner.yield_every_n_objects=128 \ --set config.rustfs.scanner.alert_excess_versions=100 \ --set config.rustfs.scanner.alert_excess_version_size=1099511627776 \ @@ -197,6 +198,7 @@ grep -q 'RUSTFS_SCANNER_IDLE_MODE: "false"' <<<"$scanner_config_output" grep -q 'RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS: "30"' <<<"$scanner_config_output" grep -q 'RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS: "2"' <<<"$scanner_config_output" grep -q 'RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS: "1"' <<<"$scanner_config_output" +grep -q 'RUSTFS_SCANNER_WALKERS_PER_DISK: "2"' <<<"$scanner_config_output" grep -q 'RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS: "128"' <<<"$scanner_config_output" grep -q 'RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS: "100"' <<<"$scanner_config_output" grep -q 'RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE: "1099511627776"' <<<"$scanner_config_output"