From 71c0efa97dde32a697fbb28b4dece06f690979c2 Mon Sep 17 00:00:00 2001 From: Hauser Date: Mon, 12 Oct 2026 15:29:10 +0800 Subject: [PATCH] fix(scanner): account remote progress and preserve dispatch limits (#8485) * fix(scanner): account remote progress and preserve dispatch limits Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(scanner): isolate remote progress by cycle Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: heihutu Co-authored-by: zhi22915 --- crates/scanner-metrics/Cargo.toml | 2 +- crates/scanner-metrics/src/metrics.rs | 191 +++++++++++++++++- crates/scanner/src/remote_scanner/stream.rs | 136 +++++++------ .../src/remote_scanner/stream/tests.rs | 92 +++++++-- crates/scanner/src/scanner.rs | 16 +- crates/scanner/src/scanner/tests.rs | 66 ++++-- crates/scanner/src/scanner_budget.rs | 8 + crates/scanner/src/scanner_io/guards.rs | 6 +- crates/scanner/src/scanner_io/io_cache.rs | 13 +- crates/scanner/src/scanner_io/io_cycle.rs | 15 +- crates/scanner/src/scanner_io/tests.rs | 58 ++++-- 11 files changed, 458 insertions(+), 145 deletions(-) diff --git a/crates/scanner-metrics/Cargo.toml b/crates/scanner-metrics/Cargo.toml index cfc0d3e19..eeedf0791 100644 --- a/crates/scanner-metrics/Cargo.toml +++ b/crates/scanner-metrics/Cargo.toml @@ -38,7 +38,7 @@ tokio = { workspace = true, features = ["sync","rt"] } [dev-dependencies] serde_json = { workspace = true } -tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] } uuid = { workspace = true, features = ["v4"] } [lib] diff --git a/crates/scanner-metrics/src/metrics.rs b/crates/scanner-metrics/src/metrics.rs index 1f018d51e..8467dbd13 100644 --- a/crates/scanner-metrics/src/metrics.rs +++ b/crates/scanner-metrics/src/metrics.rs @@ -880,6 +880,24 @@ pub struct TierRequestCount { const TIER_REQUEST_COUNTER_SLOTS: usize = TierRequestOperation::ALL.len() * TierRequestOutcome::ALL.len(); +/// Remote work belongs to one cycle; retained worker handles cannot update a later cycle. +#[derive(Default)] +pub struct RemoteScanProgress { + objects: AtomicU64, + directories: AtomicU64, +} + +impl RemoteScanProgress { + /// Record authenticated per-walk deltas without changing process-local operation counters. + pub fn record(&self, objects: u64, directories: u64) { + for (counter, delta) in [(&self.objects, objects), (&self.directories, directories)] { + if delta != 0 { + let _ = counter.try_update(Ordering::Relaxed, Ordering::Relaxed, |value| Some(value.saturating_add(delta))); + } + } + } +} + pub struct Metrics { operations: Vec, latency: Vec, @@ -889,6 +907,10 @@ pub struct Metrics { cycle_info: Arc>>, current_scan_mode: AtomicU8, current_scan_cycle_work_active: AtomicBool, + // Cycle transitions and reports lock cycle_info before remote_scan_progress. + // Cycle transitions take cycle_info before this lock; never hold it across awaits + // or while acquiring another metrics lock. Workers update their own handle directly. + remote_scan_progress: Mutex>, current_scan_cycle_objects_start: AtomicU64, current_scan_cycle_directories_start: AtomicU64, current_scan_cycle_bucket_drive_scans_start: AtomicU64, @@ -1963,6 +1985,7 @@ impl Metrics { cycle_info: Arc::new(RwLock::new(None)), current_scan_mode: AtomicU8::new(HealScanMode::Unknown as u8), current_scan_cycle_work_active: AtomicBool::new(false), + remote_scan_progress: Mutex::new(Arc::new(RemoteScanProgress::default())), current_scan_cycle_objects_start: AtomicU64::new(0), current_scan_cycle_directories_start: AtomicU64::new(0), current_scan_cycle_bucket_drive_scans_start: AtomicU64::new(0), @@ -2794,8 +2817,18 @@ impl Metrics { /// Publish a scanner cycle and its work-accounting baseline as one state transition. pub async fn start_scan_cycle_work_with_cycle(&self, cycle: CurrentCycle) -> ScanCycleWorkSnapshot { + self.start_scan_cycle_work_with_cycle_and_remote_progress(cycle, Arc::new(RemoteScanProgress::default())) + .await + } + + /// Install the cycle's remote work handle together with its cycle identity. + pub async fn start_scan_cycle_work_with_cycle_and_remote_progress( + &self, + cycle: CurrentCycle, + remote_progress: Arc, + ) -> ScanCycleWorkSnapshot { let mut current_cycle = self.cycle_info.write().await; - let snapshot = self.start_scan_cycle_work(); + let snapshot = self.begin_scan_cycle_work(remote_progress); *current_cycle = Some(cycle); snapshot } @@ -2910,6 +2943,15 @@ impl Metrics { } pub fn start_scan_cycle_work(&self) -> ScanCycleWorkSnapshot { + self.begin_scan_cycle_work(Arc::new(RemoteScanProgress::default())) + } + + fn begin_scan_cycle_work(&self, remote_progress: Arc) -> ScanCycleWorkSnapshot { + // The cycle transition serializes replacement; workers keep their own Arc. + *self + .remote_scan_progress + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = remote_progress; let snapshot = self.scan_cycle_work_snapshot(); let source_snapshot = self.scanner_source_work_values(); let replication_repair_snapshot = self.scanner_replication_repair_work_values(); @@ -3027,9 +3069,22 @@ impl Metrics { fn scan_cycle_work_since(&self, start: ScanCycleWorkSnapshot) -> ScanCycleWorkSnapshot { let current = self.scan_cycle_work_snapshot(); + let (remote_objects, remote_directories) = { + let remote = self + .remote_scan_progress + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + (remote.objects.load(Ordering::Relaxed), remote.directories.load(Ordering::Relaxed)) + }; ScanCycleWorkSnapshot { - objects_scanned: current.objects_scanned.saturating_sub(start.objects_scanned), - directories_scanned: current.directories_scanned.saturating_sub(start.directories_scanned), + objects_scanned: current + .objects_scanned + .saturating_sub(start.objects_scanned) + .saturating_add(remote_objects), + directories_scanned: current + .directories_scanned + .saturating_sub(start.directories_scanned) + .saturating_add(remote_directories), bucket_drive_scans: current.bucket_drive_scans.saturating_sub(start.bucket_drive_scans), bucket_drive_failures: current.bucket_drive_failures.saturating_sub(start.bucket_drive_failures), yield_events: current.yield_events.saturating_sub(start.yield_events), @@ -3650,14 +3705,18 @@ pub type CloseDiskFn = Arc Pin + Send>> /// Register a new disk in the global path tracker and return two callbacks: /// one to update the current path and one to deregister the disk when done. pub async fn current_path_updater(disk: &str, initial: &str) -> (UpdateCurrentPathFn, CloseDiskFn) { + register_current_path(&global_metrics().current_paths, disk, initial).await +} + +async fn register_current_path( + current_paths: &Arc>>>, + disk: &str, + initial: &str, +) -> (UpdateCurrentPathFn, CloseDiskFn) { let tracker = Arc::new(CurrentPathTracker::new(initial.to_string())); let disk_name = disk.to_string(); - global_metrics() - .current_paths - .write() - .await - .insert(disk_name.clone(), Arc::clone(&tracker)); + current_paths.write().await.insert(disk_name.clone(), Arc::clone(&tracker)); let update_fn: UpdateCurrentPathFn = { let tracker = Arc::clone(&tracker); @@ -3670,10 +3729,17 @@ pub async fn current_path_updater(disk: &str, initial: &str) -> (UpdateCurrentPa let done_fn: CloseDiskFn = { let disk = disk_name; + let current_paths = Arc::clone(current_paths); + let tracker = Arc::clone(&tracker); Arc::new(move || { let disk = disk.clone(); + let current_paths = Arc::clone(¤t_paths); + let tracker = Arc::clone(&tracker); Box::pin(async move { - global_metrics().current_paths.write().await.remove(&disk); + let mut paths = current_paths.write().await; + if paths.get(&disk).is_some_and(|current| Arc::ptr_eq(current, &tracker)) { + paths.remove(&disk); + } }) }) }; @@ -3857,6 +3923,33 @@ mod tests { assert!(!global_metrics().current_paths.read().await.contains_key(&disk)); } + #[tokio::test] + async fn stale_path_cleanup_preserves_a_replacement_scan() { + let metrics = Metrics::new(); + let (_, close_first) = register_current_path(&metrics.current_paths, "remote-node/disk", "first-bucket").await; + let (_, close_second) = register_current_path(&metrics.current_paths, "remote-node/disk", "second-bucket").await; + close_first().await; + assert_eq!(metrics.get_current_paths().await, vec!["remote-node/disk/second-bucket"]); + close_second().await; + assert!(metrics.get_current_paths().await.is_empty()); + } + + #[tokio::test] + async fn cancelled_path_guard_cleans_up_its_remote_scan() { + let metrics = Metrics::new(); + let (_, close_path) = register_current_path(&metrics.current_paths, "remote-node/disk", "bucket").await; + let guard = CloseDiskGuard::new(close_path); + assert_eq!(metrics.report().await.active_paths, vec!["remote-node/disk/bucket"]); + drop(guard); + tokio::time::timeout(Duration::from_secs(1), async { + while !metrics.get_current_paths().await.is_empty() { + tokio::task::yield_now().await; + } + }) + .await + .expect("cancelled remote path guard must remove its registration"); + } + #[tokio::test] async fn report_counts_active_scan_paths() { let metrics = Metrics::new(); @@ -5061,6 +5154,86 @@ mod tests { assert_eq!(report.usage_freshness.last_cycle_cleared_dirty_buckets, 0); } + #[test] + fn remote_scan_progress_saturates_instead_of_wrapping() { + let remote = RemoteScanProgress::default(); + remote.record(u64::MAX - 1, u64::MAX - 2); + remote.record(3, 4); + remote.record(0, 0); + assert_eq!(remote.objects.load(Ordering::Relaxed), u64::MAX); + assert_eq!(remote.directories.load(Ordering::Relaxed), u64::MAX); + } + + #[tokio::test] + async fn saturated_remote_cycle_does_not_hide_later_remote_and_local_work() { + let metrics = Metrics::new(); + let old_remote = Arc::new(RemoteScanProgress::default()); + let previous = metrics + .start_scan_cycle_work_with_cycle_and_remote_progress(CurrentCycle::default(), old_remote.clone()) + .await; + old_remote.record(u64::MAX, u64::MAX); + metrics.finish_scan_cycle_work(previous); + + let remote = Arc::new(RemoteScanProgress::default()); + let next = metrics + .start_scan_cycle_work_with_cycle_and_remote_progress(CurrentCycle::default(), remote.clone()) + .await; + old_remote.record(9, 7); + let empty = metrics.report().await; + assert_eq!(empty.current_cycle_objects_scanned, 0); + assert_eq!(empty.current_cycle_directories_scanned, 0); + remote.record(2, 1); + metrics.operations[Metric::ScanObject as usize].fetch_add(3, Ordering::Relaxed); + metrics.operations[Metric::ScanFolder as usize].fetch_add(2, Ordering::Relaxed); + let active = metrics.report().await; + assert_eq!(active.current_cycle_objects_scanned, 5); + assert_eq!(active.current_cycle_directories_scanned, 3); + metrics.finish_scan_cycle_work(next); + let finished = metrics.report().await; + assert_eq!(finished.last_cycle_objects_scanned, 5); + assert_eq!(finished.last_cycle_directories_scanned, 3); + } + + #[tokio::test] + async fn cycle_work_combines_local_and_remote_progress_without_changing_local_operations() { + let metrics = Metrics::new(); + metrics.operations[Metric::ScanObject as usize].store(10, Ordering::Relaxed); + metrics.operations[Metric::ScanFolder as usize].store(5, Ordering::Relaxed); + let remote = Arc::new(RemoteScanProgress::default()); + let start = metrics + .start_scan_cycle_work_with_cycle_and_remote_progress(CurrentCycle::default(), remote.clone()) + .await; + metrics.operations[Metric::ScanObject as usize].fetch_add(3, Ordering::Relaxed); + metrics.operations[Metric::ScanFolder as usize].fetch_add(2, Ordering::Relaxed); + remote.record(7, 4); + let active = metrics.report().await; + assert_eq!(active.current_cycle_objects_scanned, 10); + assert_eq!(active.current_cycle_directories_scanned, 6); + assert_eq!(metrics.lifetime(Metric::ScanObject), 13); + assert_eq!(metrics.lifetime(Metric::ScanFolder), 7); + metrics.finish_scan_cycle_work(start); + let finished = metrics.report().await; + assert_eq!(finished.current_cycle_objects_scanned, 0); + assert_eq!(finished.current_cycle_directories_scanned, 0); + assert_eq!(finished.last_cycle_objects_scanned, 10); + assert_eq!(finished.last_cycle_directories_scanned, 6); + + remote.record(9, 3); + let next_remote = Arc::new(RemoteScanProgress::default()); + let next = metrics + .start_scan_cycle_work_with_cycle_and_remote_progress(CurrentCycle::default(), next_remote.clone()) + .await; + let empty = metrics.report().await; + assert_eq!(empty.current_cycle_objects_scanned, 0); + assert_eq!(empty.current_cycle_directories_scanned, 0); + remote.record(11, 5); + next_remote.record(2, 1); + metrics.finish_scan_cycle_work(next); + let finished = metrics.report().await; + assert_eq!(finished.last_cycle_objects_scanned, 2); + assert_eq!(finished.last_cycle_directories_scanned, 1); + } + #[tokio::test] async fn report_includes_active_scan_cycle_work() { let metrics = Metrics::new(); diff --git a/crates/scanner/src/remote_scanner/stream.rs b/crates/scanner/src/remote_scanner/stream.rs index 3749dc5e4..8dc15dad6 100644 --- a/crates/scanner/src/remote_scanner/stream.rs +++ b/crates/scanner/src/remote_scanner/stream.rs @@ -29,7 +29,7 @@ use crate::{ use hmac::{Hmac, KeyInit, Mac}; use rustfs_credentials::try_get_rpc_token; use rustfs_heal_contracts::heal_channel::HealScanMode; -use rustfs_scanner_metrics::metrics::{Metric, Metrics}; +use rustfs_scanner_metrics::metrics::{CloseDiskGuard, Metric, Metrics, current_path_updater}; use rustfs_utils::path::path_join_buf; use serde::{Deserialize, Serialize}; use sha2::Sha256; @@ -1017,71 +1017,79 @@ pub(crate) async fn scan_remote_bucket( ))); } - let rpc_deadline = Instant::now() + NS_SCANNER_MAX_RPC_LIFETIME; - let open_deadline = (Instant::now() + NS_SCANNER_STALL_TIMEOUT).min(rpc_deadline); - let open_stream = disk.open_ns_scanner_stream(crate::NsScannerOpenRequest { - request_id, - server_epoch, - session_id, - session_sequence, - next_cycle, - leader_epoch, - body, - stall_timeout: Some(NS_SCANNER_STALL_TIMEOUT), - }); - tokio::pin!(open_stream); - let reader = tokio::select! { - _ = ctx.cancelled() => { - budget.cancel_after_unreported_remote_progress(); - return Err(RemoteScannerFailure::transport(StorageError::other( - "remote namespace scanner cancelled while opening stream", - ))); - } - result = tokio::time::timeout_at(open_deadline, &mut open_stream) => { - match result { - Err(_) => { - budget.cancel_after_unreported_remote_progress(); - return Err(RemoteScannerFailure::transport(StorageError::other( - "remote namespace scanner response headers timed out", - ))); - } - Ok(Err(err)) if err.is_internode_http_status(429) => { - return Err(RemoteScannerFailure::retry_bucket(StorageError::other(format!( - "remote namespace scanner worker rejected zero-progress work: {err}" - )))); - } - Ok(Err(err)) => { - budget.cancel_after_unreported_remote_progress(); - return Err(RemoteScannerFailure::transport(StorageError::other(format!( - "failed to open remote namespace scanner stream: {err}" - )))); - } - Ok(Ok(reader)) => reader, - } - } - }; - let authenticator = FrameAuthenticator::from_rpc_secret(request_id).map_err(|err| { - RemoteScannerFailure::transport(StorageError::other(format!( - "failed to authenticate remote namespace scanner stream: {err}" - ))) - })?; - - let stream_result = consume_remote_scanner_stream_until( - reader, - ctx, - budget.clone(), - RemoteScannerResponseExpectation { - bucket, - source: expected_source, + let disk_name = format!("{}{}", disk.host_name(), disk.path().display()); + let (_, close_path) = current_path_updater(&disk_name, bucket).await; + let mut path_guard = CloseDiskGuard::new(close_path); + let result = async { + let rpc_deadline = Instant::now() + NS_SCANNER_MAX_RPC_LIFETIME; + let open_deadline = (Instant::now() + NS_SCANNER_STALL_TIMEOUT).min(rpc_deadline); + let open_stream = disk.open_ns_scanner_stream(crate::NsScannerOpenRequest { + request_id, + server_epoch, + session_id, + session_sequence, next_cycle, - scan_plan_digest, - tier_registry_generation, - }, - authenticator, - rpc_deadline, - ) + leader_epoch, + body, + stall_timeout: Some(NS_SCANNER_STALL_TIMEOUT), + }); + tokio::pin!(open_stream); + let reader = tokio::select! { + _ = ctx.cancelled() => { + budget.cancel_after_unreported_remote_progress(); + return Err(RemoteScannerFailure::transport(StorageError::other( + "remote namespace scanner cancelled while opening stream", + ))); + } + result = tokio::time::timeout_at(open_deadline, &mut open_stream) => { + match result { + Err(_) => { + budget.cancel_after_unreported_remote_progress(); + return Err(RemoteScannerFailure::transport(StorageError::other( + "remote namespace scanner response headers timed out", + ))); + } + Ok(Err(err)) if err.is_internode_http_status(429) => { + return Err(RemoteScannerFailure::retry_bucket(StorageError::other(format!( + "remote namespace scanner worker rejected zero-progress work: {err}" + )))); + } + Ok(Err(err)) => { + budget.cancel_after_unreported_remote_progress(); + return Err(RemoteScannerFailure::transport(StorageError::other(format!( + "failed to open remote namespace scanner stream: {err}" + )))); + } + Ok(Ok(reader)) => reader, + } + } + }; + let authenticator = FrameAuthenticator::from_rpc_secret(request_id).map_err(|err| { + RemoteScannerFailure::transport(StorageError::other(format!( + "failed to authenticate remote namespace scanner stream: {err}" + ))) + })?; + + let stream_result = consume_remote_scanner_stream_until( + reader, + ctx, + budget.clone(), + RemoteScannerResponseExpectation { + bucket, + source: expected_source, + next_cycle, + scan_plan_digest, + tier_registry_generation, + }, + authenticator, + rpc_deadline, + ) + .await; + finish_remote_scanner_stream(stream_result, budget.as_ref()) + } .await; - finish_remote_scanner_stream(stream_result, budget.as_ref()) + path_guard.close().await; + result } fn finish_remote_scanner_stream( diff --git a/crates/scanner/src/remote_scanner/stream/tests.rs b/crates/scanner/src/remote_scanner/stream/tests.rs index b73b1077d..7b5e2df0b 100644 --- a/crates/scanner/src/remote_scanner/stream/tests.rs +++ b/crates/scanner/src/remote_scanner/stream/tests.rs @@ -917,18 +917,20 @@ async fn complete_terminal_frame_reconciles_progress_and_usage() { let (mut writer, reader) = tokio::io::duplex(4096); tokio::spawn(async move { let mut sequence = 0; - write_frame( - &mut writer, - &writer_auth, - &mut sequence, - &RemoteScannerFrame::progress(RemoteScannerProgress { - objects_scanned: 2, - directories_started: 1, - ..Default::default() - }), - ) - .await - .expect("progress should write"); + for _ in 0..2 { + write_frame( + &mut writer, + &writer_auth, + &mut sequence, + &RemoteScannerFrame::progress(RemoteScannerProgress { + objects_scanned: 2, + directories_started: 1, + ..Default::default() + }), + ) + .await + .expect("progress should write"); + } write_frame( &mut writer, &writer_auth, @@ -961,10 +963,26 @@ async fn complete_terminal_frame_reconciles_progress_and_usage() { ..Default::default() }, ); - let outcome = - consume_remote_scanner_stream(reader, parent, budget.clone(), "bucket", TEST_SOURCE, TEST_PLAN_DIGEST, reader_auth) - .await - .expect("stream should complete"); + let metrics = Metrics::new(); + let start = metrics + .start_scan_cycle_work_with_cycle_and_remote_progress(Default::default(), budget.remote_scan_progress()) + .await; + let outcome = consume_remote_scanner_stream_until( + reader, + parent, + budget.clone(), + RemoteScannerResponseExpectation { + bucket: "bucket", + source: TEST_SOURCE, + next_cycle: TEST_NEXT_CYCLE, + scan_plan_digest: TEST_PLAN_DIGEST, + tier_registry_generation: 0, + }, + reader_auth, + Instant::now() + NS_SCANNER_MAX_RPC_LIFETIME, + ) + .await + .expect("stream should complete"); let RemoteScannerOutcome::Complete { usage, @@ -976,6 +994,15 @@ async fn complete_terminal_frame_reconciles_progress_and_usage() { assert_eq!(usage.entry.objects, 3); assert!(pending_maintenance_work); assert_eq!(budget.progress(), (3, 2)); + let active = metrics.report().await; + assert_eq!(active.current_cycle_objects_scanned, 3); + assert_eq!(active.current_cycle_directories_scanned, 2); + assert_eq!(metrics.lifetime(Metric::ScanObject), 0); + assert_eq!(metrics.lifetime(Metric::ScanFolder), 0); + metrics.finish_scan_cycle_work(start); + let finished = metrics.report().await; + assert_eq!(finished.last_cycle_objects_scanned, 3); + assert_eq!(finished.last_cycle_directories_scanned, 2); } #[tokio::test] @@ -1271,6 +1298,39 @@ fn uncertain_stream_failure_cancels_count_budget() { assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Objects)); } +#[tokio::test] +async fn remote_progress_rejects_regression_and_counts_new_attempts_once() { + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, ScannerCycleBudgetConfig::default()); + let metrics = Metrics::new(); + let _start = metrics + .start_scan_cycle_work_with_cycle_and_remote_progress(Default::default(), budget.remote_scan_progress()) + .await; + let mut last = RemoteScannerProgress::default(); + let progress = RemoteScannerProgress { + objects_scanned: 4, + directories_started: 2, + entries_visited: 6, + }; + assert!(apply_remote_progress(&budget, &mut last, progress).expect("first progress must advance")); + assert!(!apply_remote_progress(&budget, &mut last, progress).expect("duplicate progress must be accepted")); + let regression = RemoteScannerProgress { + objects_scanned: 3, + ..progress + }; + assert!(apply_remote_progress(&budget, &mut last, regression).is_err()); + assert_eq!(last, progress); + assert_eq!(metrics.report().await.current_cycle_objects_scanned, 4); + assert_eq!(metrics.report().await.current_cycle_directories_scanned, 2); + + let mut retry = RemoteScannerProgress::default(); + apply_remote_progress(&budget, &mut retry, progress).expect("new attempt counts repeated work"); + assert_eq!(budget.progress(), (8, 4)); + let report = metrics.report().await; + assert_eq!(report.current_cycle_objects_scanned, 8); + assert_eq!(report.current_cycle_directories_scanned, 4); +} + #[test] fn directory_entry_activity_counts_as_semantic_progress() { let parent = CancellationToken::new(); diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index c41b01176..3624be128 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -51,9 +51,9 @@ use rustfs_data_usage::observed_data_usage_is_newer; use rustfs_heal_contracts::heal_channel::HealScanMode; use rustfs_lock::{NamespaceLockGuard, error::LockError}; use rustfs_scanner_metrics::metrics::{ - CurrentCycle, Metric, Metrics, ScanCyclePartialReason, ScanCycleWorkSnapshot, ScannerUsageSaveResult, ScannerWorkSource, - emit_scan_cycle_complete, emit_scan_cycle_deferred, emit_scan_cycle_partial_with_source, emit_scan_cycle_superseded, - global_metrics, + CurrentCycle, Metric, Metrics, RemoteScanProgress, ScanCyclePartialReason, ScanCycleWorkSnapshot, ScannerUsageSaveResult, + ScannerWorkSource, emit_scan_cycle_complete, emit_scan_cycle_deferred, emit_scan_cycle_partial_with_source, + emit_scan_cycle_superseded, global_metrics, }; use serde::{Deserialize, Serialize}; use sha2::{Digest as _, Sha256}; @@ -1777,7 +1777,7 @@ where let now = Instant::now(); cycle_info.started = Utc::now(); - let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; + let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone(), cycle_budget.remote_scan_progress()).await; // Refresh the storage-owned movement snapshot before reading background // heal state. A missing heal object yields an in-memory default; do not @@ -2598,9 +2598,13 @@ struct ScannerCycleMetricsGuard { } impl ScannerCycleMetricsGuard { - async fn new(cycle: CurrentCycle) -> Self { + async fn new(cycle: CurrentCycle, remote_progress: Arc) -> Self { Self { - start: Some(global_metrics().start_scan_cycle_work_with_cycle(cycle).await), + start: Some( + global_metrics() + .start_scan_cycle_work_with_cycle_and_remote_progress(cycle, remote_progress) + .await, + ), } } diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 6cc15580a..4c3c9879c 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -714,7 +714,7 @@ async fn checkpoint_runtime_handoff(store: Arc, expected_cycl assert_eq!((cycle.next, saved_epoch), (expected_cycle, epoch)); cycle.current = cycle.next; let ctx = CancellationToken::new(); - let mut metrics = ScannerCycleMetricsGuard::new(cycle.clone()).await; + let mut metrics = ScannerCycleMetricsGuard::new(cycle.clone(), Arc::new(RemoteScanProgress::default())).await; tokio::time::pause(); let budget = ScannerCycleBudget::new( &ctx, @@ -1289,7 +1289,8 @@ async fn test_mark_scan_cycle_idle_clears_published_cycle_state() { }; global_metrics().set_current_scan_mode(HealScanMode::Deep); - let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; + let mut cycle_metrics_guard = + ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await; mark_scan_cycle_idle(&mut cycle_info, &mut cycle_metrics_guard).await; @@ -1307,6 +1308,33 @@ async fn test_mark_scan_cycle_idle_clears_published_cycle_state() { global_metrics().set_cycle(None).await; } +#[tokio::test] +#[serial] +async fn scanner_cycle_metrics_guard_installs_budget_remote_progress() { + let parent = CancellationToken::new(); + let old_budget = ScannerCycleBudget::new(&parent, ScannerCycleBudgetConfig::default()); + let mut guard = ScannerCycleMetricsGuard::new(CurrentCycle::default(), old_budget.remote_scan_progress()).await; + old_budget.record_remote_progress(3, 2); + guard.finish(CurrentCycle::default()).await; + let previous = global_metrics().report().await; + assert_eq!(previous.last_cycle_objects_scanned, 3); + assert_eq!(previous.last_cycle_directories_scanned, 2); + + let budget = ScannerCycleBudget::new(&parent, ScannerCycleBudgetConfig::default()); + let mut guard = ScannerCycleMetricsGuard::new(CurrentCycle::default(), budget.remote_scan_progress()).await; + budget.record_remote_progress(2, 1); + old_budget.record_remote_progress(9, 7); + let active = global_metrics().report().await; + assert_eq!(active.current_cycle_objects_scanned, 2); + assert_eq!(active.current_cycle_directories_scanned, 1); + guard.finish(CurrentCycle::default()).await; + old_budget.record_remote_progress(11, 5); + let finished = global_metrics().report().await; + assert_eq!(finished.last_cycle_objects_scanned, 2); + assert_eq!(finished.last_cycle_directories_scanned, 1); + global_metrics().set_cycle(None).await; +} + #[tokio::test] #[serial] async fn scanner_cycle_metrics_guard_covers_published_first_cycle_lifetime() { @@ -1317,7 +1345,7 @@ async fn scanner_cycle_metrics_guard_covers_published_first_cycle_lifetime() { started: cycle_started, ..Default::default() }; - let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; + let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await; let setup_report = global_metrics().report().await; assert!(setup_report.current_cycle_active); assert_eq!(setup_report.current_cycle, 0); @@ -1343,7 +1371,7 @@ async fn scanner_cycle_metrics_guard_keeps_active_cycle_published_during_finaliz started: Utc::now(), ..Default::default() }; - let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; + let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await; cycle_info.current = 0; tokio::task::yield_now().await; @@ -1362,12 +1390,15 @@ async fn scanner_cycle_metrics_guard_keeps_active_cycle_published_during_finaliz #[tokio::test] #[serial] async fn scanner_cycle_metrics_guard_drop_clears_activity() { - let guard = ScannerCycleMetricsGuard::new(CurrentCycle { - current: 12, - next: 13, - started: Utc::now(), - ..Default::default() - }) + let guard = ScannerCycleMetricsGuard::new( + CurrentCycle { + current: 12, + next: 13, + started: Utc::now(), + ..Default::default() + }, + Arc::new(RemoteScanProgress::default()), + ) .await; assert!(global_metrics().report().await.current_cycle_active); @@ -1639,7 +1670,8 @@ async fn test_finalize_partial_scan_cycle_advances_and_persists_counter() { cycle_completed: vec![], started: Utc::now(), }; - let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; + let mut cycle_metrics_guard = + ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await; assert!(finalize_partial_scan_cycle(&ctx, store.clone(), &mut cycle_info, &mut revision, 1, &mut cycle_metrics_guard,).await); @@ -1675,7 +1707,8 @@ async fn scanner_cycle_recovers_to_newer_durable_cache_floor() { cycle_completed: vec![], started: Utc::now(), }; - let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; + let mut cycle_metrics_guard = + ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await; assert!( persist_required_scanner_cycle_floor( @@ -1715,7 +1748,8 @@ async fn scanner_cycle_rejects_invalid_cache_floor() { cycle_completed: vec![], started: Utc::now(), }; - let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; + let mut cycle_metrics_guard = + ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await; assert!( !persist_required_scanner_cycle_floor( @@ -1736,7 +1770,8 @@ async fn scanner_cycle_rejects_invalid_cache_floor() { next: 12, ..Default::default() }; - let mut max_cycle_metrics_guard = ScannerCycleMetricsGuard::new(max_cycle_info.clone()).await; + let mut max_cycle_metrics_guard = + ScannerCycleMetricsGuard::new(max_cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await; assert!( !persist_required_scanner_cycle_floor( &ctx, @@ -4600,7 +4635,8 @@ async fn test_finalize_partial_scan_cycle_reports_persist_failure() { cycle_completed: vec![], started: Utc::now(), }; - let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await; + let mut cycle_metrics_guard = + ScannerCycleMetricsGuard::new(cycle_info.clone(), Arc::new(RemoteScanProgress::default())).await; assert!(!finalize_partial_scan_cycle(&ctx, store, &mut cycle_info, &mut revision, 1, &mut cycle_metrics_guard,).await); assert_eq!(cycle_info.next, 13); diff --git a/crates/scanner/src/scanner_budget.rs b/crates/scanner/src/scanner_budget.rs index 65743439f..3a084e90c 100644 --- a/crates/scanner/src/scanner_budget.rs +++ b/crates/scanner/src/scanner_budget.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use rustfs_scanner_metrics::metrics::RemoteScanProgress; use std::sync::{ Arc, atomic::{AtomicBool, AtomicU8, AtomicU64, Ordering}, @@ -73,6 +74,7 @@ pub struct ScannerCycleBudget { entries_visited: AtomicU64, last_progress_millis: AtomicU64, cycle_state_persisted: AtomicBool, + remote_scan_progress: Arc, } impl ScannerCycleBudget { @@ -137,6 +139,7 @@ impl ScannerCycleBudget { entries_visited: AtomicU64::new(0), last_progress_millis: AtomicU64::new(0), cycle_state_persisted: AtomicBool::new(false), + remote_scan_progress: Arc::new(RemoteScanProgress::default()), }) } @@ -237,7 +240,12 @@ impl ScannerCycleBudget { } } + pub(crate) fn remote_scan_progress(&self) -> Arc { + Arc::clone(&self.remote_scan_progress) + } + pub(crate) fn record_remote_progress(&self, objects_scanned: u64, directories_started: u64) { + self.remote_scan_progress.record(objects_scanned, directories_started); if self.track_progress || self.max_objects.is_some() { let objects = saturating_fetch_add(&self.objects_scanned, objects_scanned); if self.track_progress { diff --git a/crates/scanner/src/scanner_io/guards.rs b/crates/scanner/src/scanner_io/guards.rs index bc1e0e4a2..3a5cb4fa9 100644 --- a/crates/scanner/src/scanner_io/guards.rs +++ b/crates/scanner/src/scanner_io/guards.rs @@ -546,15 +546,13 @@ pub(super) fn reset_disk_bucket_scan_gauges(pool: &str, set: &str) { record_disk_bucket_scans_active(0, pool, set); } +// Foreground pressure is handled during work by disk admission and the sleeper. +// Dispatch limits must not freeze a momentary request count for an entire cycle. pub(super) fn scanner_concurrency_limit(configured: usize, available: usize) -> usize { if available == 0 { return 0; } - if crate::workload_admission::foreground_workload_activity() > 0 { - return 1; - } - if configured == 0 { available } else { diff --git a/crates/scanner/src/scanner_io/io_cache.rs b/crates/scanner/src/scanner_io/io_cache.rs index aa51abae0..f6a9d4146 100644 --- a/crates/scanner/src/scanner_io/io_cache.rs +++ b/crates/scanner/src/scanner_io/io_cache.rs @@ -469,21 +469,22 @@ impl ScannerIOCache for SetDisks { "Scanner set assigned remote disks without namespace scanner support to coordinator-driven workers" ); } - let disk_scan_limit = scanner_budgeted_concurrency_limit( - scanner_max_concurrent_disk_scans(workers.len()), - budget.requires_serial_progress_accounting(), - ); + let dispatch_limit = scanner_max_concurrent_disk_scans(workers.len()); + 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); - debug!( + tracing::info!( target: "rustfs::scanner::io", event = EVENT_SCANNER_SET_STATE, component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_IO, + state = "disk_concurrency_budget", + reason = if serial_progress_accounting { "count_budget" } else { "configured_or_available_work" }, pool = self.pool_index, set = self.set_index, online_disks = workers.len(), + dispatch_limit, concurrency_limit = disk_scan_limit, - state = "disk_concurrency_budget", "Scanner disk concurrency budget resolved" ); let disk_scan_semaphore = Arc::new(Semaphore::new(disk_scan_limit)); diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index b77411e09..577cb1094 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -506,22 +506,23 @@ where return Ok(ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_publication_epoch(publication_epoch)); } - let set_scan_limit = scanner_budgeted_concurrency_limit( - scanner_max_concurrent_set_scans(total_results), - budget.requires_serial_progress_accounting(), - ); + let dispatch_limit = scanner_max_concurrent_set_scans(total_results); + let serial_progress_accounting = budget.requires_serial_progress_accounting(); + let set_scan_limit = scanner_budgeted_concurrency_limit(dispatch_limit, serial_progress_accounting); let bucket_failures = ScannerBucketFailureState::default(); let pending_maintenance_work = Arc::new(AtomicBool::new(false)); let cold_zero_walk_reuse_observed = Arc::new(AtomicBool::new(false)); record_set_scan_concurrency_limit(set_scan_limit); - debug!( + tracing::info!( target: "rustfs::scanner::io", event = EVENT_SCANNER_SET_STATE, component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_IO, - total_sets = total_results, - concurrency_limit = set_scan_limit, state = "concurrency_budget", + reason = if serial_progress_accounting { "count_budget" } else { "configured_or_available_work" }, + total_sets = total_results, + dispatch_limit, + concurrency_limit = set_scan_limit, "Scanner set concurrency budget resolved" ); let set_scan_semaphore = Arc::new(Semaphore::new(set_scan_limit)); diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index 08d6799e6..1a7eef5bb 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -3959,39 +3959,63 @@ fn scanner_concurrency_limit_handles_no_available_work() { #[test] #[serial] -fn scanner_concurrency_limit_yields_to_foreground_reads() { +fn scanner_concurrency_limit_is_stable_across_foreground_read_transitions() { crate::reset_foreground_read_activity_for_test(); crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); - crate::set_foreground_read_activity(8); - assert_eq!(scanner_concurrency_limit(0, 4), 1); - assert_eq!(scanner_concurrency_limit(3, 4), 1); + for foreground_activity in [0, 1, 8, 0] { + crate::set_foreground_read_activity(foreground_activity); + assert_eq!(scanner_concurrency_limit(0, 4), 4); + assert_eq!(scanner_concurrency_limit(3, 4), 3); + assert_eq!(scanner_concurrency_limit(1, 4), 1); + assert_eq!(scanner_concurrency_limit(8, 0), 0); + assert_eq!(scanner_budgeted_concurrency_limit(scanner_concurrency_limit(3, 4), true), 1); + assert_eq!(scanner_budgeted_concurrency_limit(scanner_concurrency_limit(3, 4), false), 3); + } crate::reset_foreground_read_activity_for_test(); } #[test] #[serial] -fn scanner_concurrency_limit_yields_to_shared_foreground_pressure() { +fn scanner_concurrency_limit_is_stable_across_shared_foreground_pressure() { crate::reset_foreground_read_activity_for_test(); crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); - install_scanner_workload_provider(WorkloadAdmissionRegistrySnapshot::new(vec![ - WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, AdmissionState::Open).with_counts(Some(2), None, Some(16)), - ])); - - assert_eq!(scanner_concurrency_limit(0, 4), 1); - assert_eq!(scanner_concurrency_limit(3, 4), 1); - + for (state, active) in [ + (AdmissionState::Open, Some(0)), + (AdmissionState::Open, Some(2)), + (AdmissionState::Saturated, None), + (AdmissionState::Open, Some(0)), + ] { + install_scanner_workload_provider(WorkloadAdmissionRegistrySnapshot::new(vec![ + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(active, None, Some(16)), + ])); + assert_eq!(scanner_concurrency_limit(0, 4), 4); + assert_eq!(scanner_concurrency_limit(3, 4), 3); + } crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); } #[test] #[serial] -fn scanner_concurrency_limit_yields_to_streaming_reads() { +fn scanner_dispatch_limits_preserve_configured_caps_during_streaming_reads() { crate::reset_foreground_read_activity_for_test(); crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); - let _guard = crate::ForegroundReadGuard::new(); - - assert_eq!(scanner_concurrency_limit(0, 4), 1); - assert_eq!(scanner_concurrency_limit(3, 4), 1); + temp_env::with_vars( + [ + (ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, Some("3")), + (ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, Some("2")), + ], + || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); + for streaming in [false, true, false] { + let _guard = streaming.then(crate::ForegroundReadGuard::new); + assert_eq!(scanner_max_concurrent_set_scans(4), 3); + assert_eq!(scanner_max_concurrent_disk_scans(4), 2); + assert_eq!(scanner_budgeted_concurrency_limit(scanner_max_concurrent_set_scans(4), true), 1); + assert_eq!(scanner_budgeted_concurrency_limit(scanner_max_concurrent_disk_scans(4), true), 1); + } + }, + ); + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } #[test]