From 62e66baf892fcb35c9172317acf7eb825b48ac9b Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 3 Sep 2026 20:31:59 +0800 Subject: [PATCH] feat(scanner): add storage seam (#7105) --- ARCHITECTURE.md | 2 +- Cargo.lock | 5 +- Cargo.toml | 2 - crates/scanner-contracts/Cargo.toml | 35 - crates/scanner-contracts/src/lib.rs | 13 - crates/scanner/Cargo.toml | 1 + crates/scanner/src/lib.rs | 20 +- crates/scanner/src/scanner.rs | 150 +++-- crates/scanner/src/scanner/activity.rs | 21 +- crates/scanner/src/scanner/backlog.rs | 46 +- crates/scanner/src/scanner/heal_info.rs | 17 +- crates/scanner/src/scanner/tests.rs | 3 +- crates/scanner/src/scanner_io.rs | 21 +- crates/scanner/src/scanner_io/io_cycle.rs | 774 +++++++++++----------- crates/scanner/src/storage_api.rs | 296 ++++++++- docs/architecture/crate-boundaries.md | 4 +- 16 files changed, 850 insertions(+), 560 deletions(-) delete mode 100644 crates/scanner-contracts/Cargo.toml delete mode 100644 crates/scanner-contracts/src/lib.rs diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index da8f386a0..189de36bb 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -92,7 +92,7 @@ refactors. | Domain | Current workspace crates | Responsibility | |--------|--------------------------|----------------| -| Foundation | `checksums`, `common`, `config`, `data-usage`, `heal-contracts`, `scanner-contracts`, `scanner-metrics`, `utils` | Shared configuration, data-usage models, heal/scanner domain contracts and telemetry types, utilities, and checksums. | +| Foundation | `checksums`, `common`, `config`, `data-usage`, `heal-contracts`, `scanner-metrics`, `utils` | Shared configuration, data-usage models, heal domain contracts, scanner telemetry types, utilities, and checksums. | | I/O and storage | `concurrency`, `ecstore`, `filemeta`, `heal`, `io-core`, `io-metrics`, `lifecycle`, `lock`, `object-capacity`, `object-data-cache`, `replication`, `rio`, `rio-v2`, `s3-client`, `scanner`, `storage-api` | Erasure-coded object storage, metadata, recovery, lifecycle, replication, locking, cache, I/O pipelines, and the engine-side S3 client for remote tier/transition targets. | | Security and identity | `credentials`, `crypto`, `iam`, `keystone`, `kms`, `policy`, `security-governance`, `signer`, `tls-runtime`, `trusted-proxies` | Credentials, authentication, authorization, encryption, key management, TLS, and security contracts. | | Protocols and contracts | `extension-schema`, `madmin`, `protos`, `protocols`, `s3-ops`, `s3-types`, `s3select-api`, `s3select-query` | Admin, inter-node, S3, S3 Select, and optional protocol contracts. | diff --git a/Cargo.lock b/Cargo.lock index e9195be56..2aba26999 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10681,6 +10681,7 @@ dependencies = [ "rustfs-ecstore", "rustfs-filemeta", "rustfs-heal-contracts", + "rustfs-lifecycle", "rustfs-lock", "rustfs-s3-types", "rustfs-scanner-metrics", @@ -10702,10 +10703,6 @@ dependencies = [ "uuid", ] -[[package]] -name = "rustfs-scanner-contracts" -version = "1.0.0-rc.5" - [[package]] name = "rustfs-scanner-metrics" version = "1.0.0-rc.5" diff --git a/Cargo.toml b/Cargo.toml index fc7b44766..644e5b08a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -52,7 +52,6 @@ members = [ "crates/s3select-api", # S3 Select API interface "crates/s3select-query", # S3 Select query engine "crates/scanner", # Scanner for data integrity checks and health monitoring - "crates/scanner-contracts", # Scanner storage and wire contracts "crates/scanner-metrics", # Scanner metrics and cycle telemetry "crates/security-governance", # Security governance contracts "crates/extension-schema", # Extension schema contracts @@ -94,7 +93,6 @@ redundant_clone = "warn" rustfs = { path = "./rustfs", version = "1.0.0-rc.5" } rustfs-heal = { path = "crates/heal", version = "1.0.0-rc.5" } rustfs-heal-contracts = { path = "crates/heal-contracts", version = "1.0.0-rc.5" } -rustfs-scanner-contracts = { path = "crates/scanner-contracts", version = "1.0.0-rc.5" } rustfs-scanner-metrics = { path = "crates/scanner-metrics", version = "1.0.0-rc.5" } rustfs-audit = { path = "crates/audit", version = "1.0.0-rc.5" } rustfs-checksums = { path = "crates/checksums", version = "1.0.0-rc.5" } diff --git a/crates/scanner-contracts/Cargo.toml b/crates/scanner-contracts/Cargo.toml deleted file mode 100644 index ecfcd5f2e..000000000 --- a/crates/scanner-contracts/Cargo.toml +++ /dev/null @@ -1,35 +0,0 @@ -# Copyright 2024 RustFS Team -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -[package] -name = "rustfs-scanner-contracts" -version.workspace = true -edition.workspace = true -license.workspace = true -repository.workspace = true -rust-version.workspace = true -homepage.workspace = true -description = "Scanner storage and wire contracts shared by the scanner and storage engine." -keywords = ["scanner", "contracts", "rustfs", "Minio"] -categories = ["web-programming", "development-tools", "data-structures"] - -[lints] -workspace = true - -[dependencies] - -[dev-dependencies] - -[lib] -doctest = false diff --git a/crates/scanner-contracts/src/lib.rs b/crates/scanner-contracts/src/lib.rs deleted file mode 100644 index 6238cfff4..000000000 --- a/crates/scanner-contracts/src/lib.rs +++ /dev/null @@ -1,13 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. diff --git a/crates/scanner/Cargo.toml b/crates/scanner/Cargo.toml index 16b15f9ba..7061a5e0f 100644 --- a/crates/scanner/Cargo.toml +++ b/crates/scanner/Cargo.toml @@ -75,6 +75,7 @@ rustfs-config = { workspace = true, features = ["server-config-model"] } rustfs-common = { workspace = true } rustfs-concurrency = { workspace = true } rustfs-heal-contracts = { workspace = true } +rustfs-lifecycle = { workspace = true } rustfs-scanner-metrics = { workspace = true } rustfs-credentials = { workspace = true } rustfs-utils = { workspace = true } diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 8ea602b37..393392a01 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -43,7 +43,7 @@ use storage_api::owner::{ ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, - ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info, + ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config, scanner_replication_config_for_lifecycle_eval, @@ -687,10 +687,6 @@ pub(crate) async fn scanner_is_erasure() -> bool { ecstore_is_erasure().await } -pub(crate) async fn scanner_is_erasure_sd() -> bool { - ecstore_is_erasure_sd().await -} - pub(crate) async fn scanner_disk_is_online(disk: &Disk) -> bool { EcstoreDiskAPI::is_online(disk).await } @@ -977,7 +973,7 @@ impl ScannerConfigObjectDelete for ECStore { } async fn scanner_data_usage_publication_admission(&self) -> Option { - let (read_guard, epoch) = self.scanner_data_usage_publication_admission_guard().await?; + let (read_guard, epoch) = ECStore::scanner_data_usage_publication_admission_guard(self).await?; Some(ScannerDataUsagePublicationAdmission::fenced(read_guard, epoch)) } @@ -987,7 +983,7 @@ impl ScannerConfigObjectDelete for ECStore { safe_deadline: tokio::time::Instant, remote_lease_tokens: Vec, ) -> Option { - self.scanner_data_usage_publication_commit_scope(expected_movement_epoch, safe_deadline, remote_lease_tokens) + ECStore::scanner_data_usage_publication_commit_scope(self, expected_movement_epoch, safe_deadline, remote_lease_tokens) .await } @@ -998,7 +994,8 @@ impl ScannerConfigObjectDelete for ECStore { remote_lease_tokens: Vec, lease_release_safe: Arc, ) -> Option { - self.scanner_data_usage_publication_commit_scope_with_release_flag( + ECStore::scanner_data_usage_publication_commit_scope_with_release_flag( + self, expected_movement_epoch, safe_deadline, remote_lease_tokens, @@ -1020,7 +1017,7 @@ impl ScannerConfigObjectDelete for SetDisks { } async fn scanner_data_usage_publication_admission(&self) -> Option { - let (read_guard, epoch) = self.scanner_data_usage_publication_admission_guard().await?; + let (read_guard, epoch) = SetDisks::scanner_data_usage_publication_admission_guard(self).await?; Some(ScannerDataUsagePublicationAdmission::fenced(read_guard, epoch)) } @@ -1030,7 +1027,7 @@ impl ScannerConfigObjectDelete for SetDisks { safe_deadline: tokio::time::Instant, remote_lease_tokens: Vec, ) -> Option { - self.scanner_data_usage_publication_commit_scope(expected_movement_epoch, safe_deadline, remote_lease_tokens) + SetDisks::scanner_data_usage_publication_commit_scope(self, expected_movement_epoch, safe_deadline, remote_lease_tokens) .await } @@ -1041,7 +1038,8 @@ impl ScannerConfigObjectDelete for SetDisks { remote_lease_tokens: Vec, lease_release_safe: Arc, ) -> Option { - self.scanner_data_usage_publication_commit_scope_with_release_flag( + SetDisks::scanner_data_usage_publication_commit_scope_with_release_flag( + self, expected_movement_epoch, safe_deadline, remote_lease_tokens, diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 43740e45d..62a3d0527 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -33,9 +33,8 @@ use crate::runtime_config::{ use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig, ScannerCycleBudgetReason}; use crate::scanner_folder::{data_usage_update_dir_cycles, heal_object_select_prob}; use crate::scanner_io::{ - ScannerCycleDeferReason, ScannerCycleResult, ScannerCycleStatus, ScannerIOCycle, dirty_usage_bucket_notified, - dirty_usage_buckets_pending, dirty_usage_generation, scanner_dirty_usage_state, scanner_maintenance_changed, - scanner_maintenance_generation, + ScannerCycleDeferReason, ScannerCycleResult, ScannerCycleStatus, dirty_usage_bucket_notified, dirty_usage_buckets_pending, + dirty_usage_generation, scanner_dirty_usage_state, scanner_maintenance_changed, scanner_maintenance_generation, }; use crate::sleeper::{SCANNER_SLEEPER, set_scanner_default_speed}; use crate::{DataUsageInfo, ScannerActivityGuard, ScannerError, ScannerRuntimeGuard}; @@ -66,17 +65,18 @@ use tokio_util::task::AbortOnDropHandle; use tracing::{debug, error, info, instrument, warn}; use uuid::Uuid; +use crate::storage_api::ScannerStorage; use crate::storage_api::scan::{ - BucketOperations, BucketOptions, NamespaceLocking as _, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, - SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION, + BucketOptions, NamespaceLocking as _, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, + SCANNER_ACTIVITY_PROTOCOL_VERSION, }; use crate::{ ECStore, EcstoreError, RUSTFS_META_BUCKET, SCANNER_PUBLICATION_EPOCH_CHANGED, ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _, delete_config_with_publication_admission_for_epoch, get_lifecycle_config, get_replication_config, invalidate_admin_data_usage_snapshot_cache, invalidate_data_usage_snapshot_cache, read_config, replace_bucket_usage_memory_from_info, save_config, save_config_shared_with_preconditions_and_lease_fence_and_scope, - save_config_with_preconditions, save_config_with_publication_admission_for_epoch, scanner_is_erasure_sd, - scanner_publication_admission_for_epoch, scanner_publication_epoch, scanner_publication_epoch_changed, + save_config_with_preconditions, save_config_with_publication_admission_for_epoch, scanner_publication_admission_for_epoch, + scanner_publication_epoch, scanner_publication_epoch_changed, }; const LOG_COMPONENT_SCANNER: &str = "scanner"; @@ -211,12 +211,18 @@ async fn notify_scanner_startup_observed_for_test() { } #[cfg(test)] -fn scanner_observed_probe_store_key(storeapi: &Arc) -> usize { - Arc::as_ptr(storeapi).cast::<()>() as usize +fn scanner_observed_probe_store_key(storeapi: &Arc) -> usize +where + S: ScannerStorage, +{ + storeapi.scanner_observed_probe_store_key() } #[cfg(test)] -fn notify_scanner_runtime_observed_for_test(storeapi: &Arc, observation: ScannerPauseBacklogObservation) { +fn notify_scanner_runtime_observed_for_test(storeapi: &Arc, observation: ScannerPauseBacklogObservation) +where + S: ScannerStorage, +{ if let Some(probe) = SCANNER_RUNTIME_OBSERVED_PROBE .lock() .expect("scanner runtime observed probe should not be poisoned") @@ -788,7 +794,10 @@ async fn sync_data_usage_backup_from_primary_for_epoch_and_lease_and_fence_and_s )) } -async fn persisted_usage_cache_is_cold_for_startup(storeapi: &Arc) -> bool { +async fn persisted_usage_cache_is_cold_for_startup(storeapi: &Arc) -> bool +where + S: ScannerObjectIO + ScannerConfigObjectDelete, +{ let Some(data) = (match read_data_usage_config_for_startup(storeapi).await { Ok(data) => data, Err(err) => { @@ -862,7 +871,10 @@ async fn persisted_usage_cache_is_cold_for_startup(storeapi: &Arc) -> b } } -async fn initial_scanner_startup_usage_state(storeapi: &Arc) -> (bool, bool) { +async fn initial_scanner_startup_usage_state(storeapi: &Arc) -> (bool, bool) +where + S: ScannerStorage, +{ let has_buckets = match storeapi .list_bucket(&BucketOptions { no_metadata: true, @@ -933,6 +945,13 @@ fn prepare_cycle_for_usage_floor_bootstrap( } pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { + init_data_scanner_with_storage(ctx, storeapi).await; +} + +async fn init_data_scanner_with_storage(ctx: CancellationToken, storeapi: Arc) +where + S: ScannerStorage, +{ let (startup_features, startup_maintenance_generation) = configure_scanner_defaults(&ctx, &storeapi).await; // Force init global sleeper so config is read once at startup. let _ = &*SCANNER_SLEEPER; @@ -1114,7 +1133,10 @@ fn single_disk_default_speed() -> ScannerSpeed { ScannerSpeed::Default } -async fn detect_scanner_maintenance_features(storeapi: &Arc) -> ScannerMaintenanceFeatures { +async fn detect_scanner_maintenance_features(storeapi: &Arc) -> ScannerMaintenanceFeatures +where + S: ScannerStorage, +{ let mut features = ScannerMaintenanceFeatures::default(); let buckets = match storeapi .list_bucket(&BucketOptions { @@ -1194,7 +1216,7 @@ async fn detect_scanner_maintenance_features(storeapi: &Arc) -> Scanner async fn detect_stable_scanner_maintenance_features( ctx: &CancellationToken, - storeapi: &Arc, + storeapi: &Arc, ) -> Option<(ScannerMaintenanceFeatures, u64)> { detect_stable_scanner_maintenance_features_with( ctx, @@ -1259,7 +1281,7 @@ where async fn configure_scanner_defaults( ctx: &CancellationToken, - storeapi: &Arc, + storeapi: &Arc, ) -> (ScannerMaintenanceFeatures, Option) { if storeapi.setup_is_erasure_sd().await { let (features, maintenance_generation) = detect_stable_scanner_maintenance_features(ctx, storeapi) @@ -1531,27 +1553,33 @@ async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle, cycle_metrics_guard } #[cfg(test)] -async fn run_data_scanner_cycle( +async fn run_data_scanner_cycle( ctx: &CancellationToken, - storeapi: &Arc, + storeapi: &Arc, cycle_info: &mut CurrentCycle, cycle_revision: &mut DataUsageCacheRevision, leader_epoch: u64, -) -> ScannerCycleOutcome { +) -> ScannerCycleOutcome +where + S: ScannerStorage, +{ let cycle_budget = ScannerCycleBudget::new(ctx, scanner_cycle_budget_config()); run_data_scanner_cycle_with_budget(ctx, storeapi, cycle_info, cycle_revision, leader_epoch, cycle_budget).await } #[instrument(skip_all)] #[hotpath::measure] -async fn run_data_scanner_cycle_with_budget( +async fn run_data_scanner_cycle_with_budget( ctx: &CancellationToken, - storeapi: &Arc, + storeapi: &Arc, cycle_info: &mut CurrentCycle, cycle_revision: &mut DataUsageCacheRevision, leader_epoch: u64, cycle_budget: Arc, -) -> ScannerCycleOutcome { +) -> ScannerCycleOutcome +where + S: ScannerStorage, +{ let _activity_guard = ScannerActivityGuard::new(); if let Err(err) = refresh_scanner_runtime_config_from_global() { warn!( @@ -1646,12 +1674,11 @@ async fn run_data_scanner_cycle_with_budget( // scanner aggregate. Hold only the short storage-owned admission guard // across this metadata read; the full bucket scan runs after it is // released and carries the captured epoch forward. - let Some((baseline_publication_guard, baseline_publication_epoch)) = - storeapi.scanner_data_usage_publication_admission_guard().await - else { + let Some(baseline_publication_guard) = storeapi.scanner_data_usage_publication_admission().await else { mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await; return ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement); }; + let baseline_publication_epoch = baseline_publication_guard.epoch(); let usage_persist_baseline_result = read_data_usage_persist_baseline(storeapi.clone()).await; drop(baseline_publication_guard); let usage_persist_baseline = match usage_persist_baseline_result { @@ -1676,17 +1703,16 @@ async fn run_data_scanner_cycle_with_budget( let (sender, receiver) = mpsc::channel::(1); let done_cycle = Metrics::time(Metric::ScanCycle); - let scan_result = storeapi - .clone() - .nsscanner_with_status( - cycle_budget.token(), - cycle_budget.clone(), - sender, - cycle_info.current, - leader_epoch, - scan_mode, - ) - .await; + let scan_result = crate::scanner_io::nsscanner_with_storage_status( + storeapi.as_ref(), + cycle_budget.token(), + cycle_budget.clone(), + sender, + cycle_info.current, + leader_epoch, + scan_mode, + ) + .await; let publication_defer_reason = match &scan_result { Ok(result) if result @@ -1725,7 +1751,7 @@ async fn run_data_scanner_cycle_with_budget( let mut remote_publication_leases = None; let remote_lease_defer_reason = if remote_publication_lease_targets.is_empty() { None - } else if let Some(notification_system) = storeapi.notification_system() { + } else if let Some(notification_system) = storeapi.scanner_notification_system() { let publication_proof_ctx = cycle_budget.token(); let lease_result = await_scanner_publication_proof( &publication_proof_ctx, @@ -2089,7 +2115,7 @@ async fn run_data_scanner_cycle_with_budget( finalize_scanner_cycle_result(scan_cycle_result, usage_persist_outcome); let remote_dirty_usage_pending = if remote_dirty_usage_acknowledgements.is_empty() { false - } else if let Some(notification_system) = storeapi.notification_system() { + } else if let Some(notification_system) = storeapi.scanner_notification_system() { let acknowledgement_count = remote_dirty_usage_acknowledgements.len(); let acknowledgements = remote_dirty_usage_acknowledgements .into_iter() @@ -2330,11 +2356,21 @@ impl Drop for ScannerCycleMetricsGuard { } pub async fn run_data_scanner(ctx: CancellationToken, storeapi: Arc) -> Result<(), ScannerError> { + run_data_scanner_with_storage(ctx, storeapi).await +} + +async fn run_data_scanner_with_storage(ctx: CancellationToken, storeapi: Arc) -> Result<(), ScannerError> +where + S: ScannerStorage, +{ let (maintenance_features, maintenance_generation) = configure_scanner_defaults(&ctx, &storeapi).await; run_data_scanner_with_maintenance_state(ctx, storeapi, maintenance_features, maintenance_generation).await } -async fn current_scanner_pause_backlog_observation(storeapi: &Arc) -> ScannerPauseBacklogObservation { +async fn current_scanner_pause_backlog_observation(storeapi: &Arc) -> ScannerPauseBacklogObservation +where + S: ScannerStorage, +{ let now_unix_secs = scanner_pause_backlog_now(); let pause = storeapi.scanner_data_movement_pause_status().await; let metrics = global_metrics().report().await; @@ -2358,12 +2394,15 @@ async fn current_scanner_pause_backlog_observation(storeapi: &Arc) -> S } } -async fn wait_for_scanner_data_movement_resume( +async fn wait_for_scanner_data_movement_resume( ctx: &CancellationToken, - storeapi: &Arc, + storeapi: &Arc, guard: &NamespaceLockGuard, - pause_backlog: &mut ScannerPauseBacklogController, -) -> bool { + pause_backlog: &mut ScannerPauseBacklogController, +) -> bool +where + S: ScannerStorage, +{ loop { let observation = current_scanner_pause_backlog_observation(storeapi).await; pause_backlog.observe(observation).await; @@ -2386,12 +2425,14 @@ async fn wait_for_scanner_data_movement_resume( } } -async fn finish_scanner_pause_backlog_cycle( - pause_backlog: &mut ScannerPauseBacklogController, - storeapi: &Arc, +async fn finish_scanner_pause_backlog_cycle( + pause_backlog: &mut ScannerPauseBacklogController, + storeapi: &Arc, attempt: ScannerPauseBacklogAttemptDecision, outcome: ScannerCycleOutcome, -) { +) where + S: ScannerStorage, +{ let observation = current_scanner_pause_backlog_observation(storeapi).await; if let ScannerPauseBacklogAttemptDecision::Tracked(serial) = attempt { pause_backlog.finish_attempt(serial, outcome, observation).await; @@ -2402,12 +2443,15 @@ async fn finish_scanner_pause_backlog_cycle( notify_scanner_runtime_observed_for_test(storeapi, observation); } -async fn run_data_scanner_with_maintenance_state( +async fn run_data_scanner_with_maintenance_state( ctx: CancellationToken, - storeapi: Arc, + storeapi: Arc, mut maintenance_features: ScannerMaintenanceFeatures, mut maintenance_generation_seen: Option, -) -> Result<(), ScannerError> { +) -> Result<(), ScannerError> +where + S: ScannerStorage, +{ reset_scanner_cycle_schedule(); // Acquire leader lock (write lock) to ensure only one scanner runs let mut guard = match storeapi.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock").await { @@ -3162,10 +3206,10 @@ impl Drop for ScannerScanModeGuard { } } -async fn final_data_usage_publication_defer_reason( - storeapi: &ECStore, - status: ScannerCycleStatus, -) -> Option { +async fn final_data_usage_publication_defer_reason(storeapi: &S, status: ScannerCycleStatus) -> Option +where + S: ScannerStorage, +{ match status { ScannerCycleStatus::Complete | ScannerCycleStatus::Superseded => { if storeapi.scanner_data_usage_publication_blocked().await { diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index e31f6c285..815679c01 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/crates/scanner/src/scanner/activity.rs @@ -13,6 +13,7 @@ // limitations under the License. /// Cycle wake/backoff policy and scanner activity observation (probing, generations, topology digest). use super::*; +use crate::storage_api::ScannerStorage; use crate::storage_api::scan::SCANNER_ACTIVITY_V6_PROTOCOL_VERSION; #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -868,14 +869,17 @@ pub(super) fn apply_scanner_activity_probe_result( } } -pub(super) async fn observe_scanner_activity( - storeapi: &Arc, +pub(super) async fn observe_scanner_activity( + storeapi: &Arc, distributed: bool, activity_seen: &mut Option, -) -> ScannerActivityObservation { +) -> ScannerActivityObservation +where + S: ScannerStorage, +{ let had_baseline = activity_seen.is_some(); let (observation, probe_error) = - apply_scanner_activity_probe_result(activity_seen, probe_scanner_activity(storeapi, distributed).await); + apply_scanner_activity_probe_result(activity_seen, probe_scanner_activity(storeapi.as_ref(), distributed).await); if let Some(err) = probe_error { log_scanner_activity_probe_error(had_baseline, &err); } @@ -978,8 +982,11 @@ pub(super) fn record_scanner_activity_instance( Ok(()) } -pub(crate) async fn probe_scanner_activity(storeapi: &ECStore, distributed: bool) -> Result { - let topology_digest = scanner_topology_digest(storeapi); +pub(crate) async fn probe_scanner_activity(storeapi: &S, distributed: bool) -> Result +where + S: ScannerStorage, +{ + let topology_digest = storeapi.scanner_topology_digest(); let (data_movement_active, publication_blocked, movement_generation) = storeapi.scanner_data_movement_activity().await; let namespace_generation = storeapi.scanner_namespace_mutation_generation(); let maintenance_generation = scanner_maintenance_generation(); @@ -1013,7 +1020,7 @@ pub(crate) async fn probe_scanner_activity(storeapi: &ECStore, distributed: bool } let notification_system = storeapi - .notification_system() + .scanner_notification_system() .ok_or_else(|| "notification system is not initialized".to_string())?; let peers = notification_system .scanner_activity_snapshots() diff --git a/crates/scanner/src/scanner/backlog.rs b/crates/scanner/src/scanner/backlog.rs index 2f160a42a..8af347340 100644 --- a/crates/scanner/src/scanner/backlog.rs +++ b/crates/scanner/src/scanner/backlog.rs @@ -21,6 +21,7 @@ use super::ScannerCycleOutcome; use crate::data_usage_define::DataUsageCacheRevision; +use crate::storage_api::ScannerStorage; use crate::storage_api::owner::ObjectIO as _; use crate::{ BUCKET_META_PREFIX, ECStore, EcstoreError, RUSTFS_META_BUCKET, ScannerObjectOptions, SetDisks, save_config_with_preconditions, @@ -1141,7 +1142,10 @@ fn select_scanner_pause_backlog_replicas(replicas: Vec) -> Result { +async fn load_scanner_pause_backlog(storeapi: Arc) -> Result +where + S: ScannerStorage, +{ let writable = storeapi.scanner_pause_backlog_writable_set_disks().await; if writable.is_empty() { return Err("scanner pause backlog has no surviving storage replicas".to_string()); @@ -1150,11 +1154,14 @@ async fn load_scanner_pause_backlog(storeapi: Arc) -> Result, +async fn write_scanner_pause_backlog_record( + storeapi: Arc, loaded: &LoadedScannerPauseBacklog, record: ScannerPauseBacklogReplicaRecord, -) -> Result<(), String> { +) -> Result<(), String> +where + S: ScannerStorage, +{ let data = serde_json::to_vec(&record).map_err(|err| format!("failed to encode scanner pause backlog: {err}"))?; if data.len() > usize::try_from(MAX_SCANNER_PAUSE_BACKLOG_BYTES).unwrap_or(usize::MAX) { return Err("scanner pause backlog exceeds its size bound".to_string()); @@ -1219,10 +1226,13 @@ async fn write_scanner_pause_backlog_record( Ok(()) } -async fn stabilize_scanner_pause_backlog( - storeapi: Arc, +async fn stabilize_scanner_pause_backlog( + storeapi: Arc, loaded: &LoadedScannerPauseBacklog, -) -> Result { +) -> Result +where + S: ScannerStorage, +{ let committed = loaded.authoritative_commit.clone().unwrap_or_else(|| { ScannerPauseBacklogCommitRecord::new(loaded.ledger.clone(), scanner_pause_backlog_replica_ids(&loaded.replicas)) }); @@ -1253,11 +1263,14 @@ fn committed_scanner_pause_backlog_pending_reload( } } -async fn persist_scanner_pause_backlog( - storeapi: Arc, +async fn persist_scanner_pause_backlog( + storeapi: Arc, loaded: &LoadedScannerPauseBacklog, ledger: ScannerPauseBacklogLedger, -) -> Result { +) -> Result +where + S: ScannerStorage, +{ let mut base = if loaded.requires_reload { let reloaded = load_scanner_pause_backlog(storeapi.clone()).await?; if reloaded.ledger != loaded.ledger { @@ -1288,15 +1301,18 @@ async fn persist_scanner_pause_backlog( } } -pub(super) struct ScannerPauseBacklogController { - storeapi: Arc, +pub(super) struct ScannerPauseBacklogController { + storeapi: Arc, loaded: LoadedScannerPauseBacklog, persistence_disabled: bool, persistence_retry_at_unix_secs: u64, } -impl ScannerPauseBacklogController { - pub(super) async fn claim(storeapi: Arc, now: u64) -> Result { +impl ScannerPauseBacklogController +where + S: ScannerStorage, +{ + pub(super) async fn claim(storeapi: Arc, now: u64) -> Result { let loaded = load_scanner_pause_backlog(storeapi.clone()).await?; let mut ledger = loaded.ledger.clone(); ledger.claim_writer(now)?; @@ -1313,7 +1329,7 @@ impl ScannerPauseBacklogController { Ok(controller) } - pub(super) fn unavailable(storeapi: Arc, error: String, now: u64) -> Self { + pub(super) fn unavailable(storeapi: Arc, error: String, now: u64) -> Self { set_runtime_error(Some(error)); let loaded = LoadedScannerPauseBacklog { ledger: ScannerPauseBacklogLedger::default(), diff --git a/crates/scanner/src/scanner/heal_info.rs b/crates/scanner/src/scanner/heal_info.rs index 55e2a0dba..f7e3f61f9 100644 --- a/crates/scanner/src/scanner/heal_info.rs +++ b/crates/scanner/src/scanner/heal_info.rs @@ -62,9 +62,12 @@ pub async fn read_background_heal_info(storeapi: Arc) -> BackgroundHeal /// Read background healing information together with the movement epoch that /// fenced the read. The epoch must be reused by the matching cycle update so a /// missing-object default cannot be committed across a movement transition. -pub(super) async fn read_background_heal_info_with_epoch(storeapi: Arc) -> BackgroundHealInfoRead { +pub(super) async fn read_background_heal_info_with_epoch(storeapi: Arc) -> BackgroundHealInfoRead +where + S: ScannerStorage, +{ // Skip for ErasureSD setup - if scanner_is_erasure_sd().await { + if storeapi.setup_is_erasure_sd().await { return BackgroundHealInfoRead { info: BackgroundHealInfo::default(), expected_epoch: None, @@ -136,13 +139,15 @@ pub async fn save_background_heal_info(storeapi: Arc, info: BackgroundH save_background_heal_info_for_epoch(storeapi, info, None).await; } -pub(super) async fn save_background_heal_info_for_epoch( - storeapi: Arc, +pub(super) async fn save_background_heal_info_for_epoch( + storeapi: Arc, info: BackgroundHealInfo, expected_epoch: Option, -) { +) where + S: ScannerStorage, +{ // Skip for ErasureSD setup - if scanner_is_erasure_sd().await { + if storeapi.setup_is_erasure_sd().await { return; } diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index e4c836c2e..d9abe98e6 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -15,6 +15,7 @@ use super::heal_info::{classify_background_heal_read_error, decode_background_heal_info}; use super::*; use crate::EcstoreResult; +use crate::storage_api::scan::BucketOperations as _; use crate::{ DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_CACHE_KEY_FORMAT, DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, DataUsageCachePrepareOutcome, DataUsageCacheSource, DataUsageEntry, DataUsageScanPlanDigest, Endpoint, EndpointServerPools, @@ -232,7 +233,7 @@ async fn restarted_main_loop_completes_durable_pause_backlog_catch_up() { } #[tokio::test] -#[serial_test::serial(scanner_runtime_env)] +#[serial] async fn running_main_loop_catches_up_pause_cleared_after_startup_observe() { temp_env::async_with_vars([(ENV_SCANNER_CYCLE, Some("1")), (ENV_SCANNER_START_DELAY_SECS, Some("0"))], async { crate::runtime_config::refresh_scanner_runtime_config_for_tests(); diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 9388f74d5..79af335cf 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -54,15 +54,16 @@ use tokio_util::task::AbortOnDropHandle; use tracing::{debug, error, warn}; use crate::ScannerObjectInfo as ObjectInfo; +use crate::storage_api::ScannerStorage; use crate::storage_api::scan::NamespaceLocking as _; use crate::storage_api::scanner_io::{BucketInfo, BucketOptions}; use crate::{ BucketTargetSys, BucketVersioningSys, Disk, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result, - RUSTFS_META_BUCKET, ReplicationConfig, STORAGE_FORMAT_FILE, ScannerConfigObjectDelete as _, ScannerDiskExt as _, - ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, - begin_tier_registry_cycle, complete_tier_registry_cycle, enqueue_runtime_free_version, get_lifecycle_config, - get_object_lock_config, get_replication_config, runtime_tier_names, runtime_tier_registry_for_cycle, - scanner_publication_admission_for_epoch, scanner_publication_epoch, storageclass, + RUSTFS_META_BUCKET, ReplicationConfig, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _, + ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, begin_tier_registry_cycle, + complete_tier_registry_cycle, enqueue_runtime_free_version, get_lifecycle_config, get_object_lock_config, + get_replication_config, runtime_tier_names, runtime_tier_registry_for_cycle, scanner_publication_admission_for_epoch, + scanner_publication_epoch, storageclass, }; pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file"; @@ -315,11 +316,14 @@ enum ScannerCycleActivityStatus { Unverified, } -async fn scanner_cycle_activity_status( - store: &ECStore, +async fn scanner_cycle_activity_status( + store: &S, distributed: bool, before: &crate::scanner::ScannerActivitySnapshot, -) -> (ScannerCycleActivityStatus, Vec<(String, String, u64)>) { +) -> (ScannerCycleActivityStatus, Vec<(String, String, u64)>) +where + S: ScannerStorage, +{ match crate::scanner::probe_scanner_activity(store, distributed).await { Ok(after) => { let status = if after == *before { @@ -728,6 +732,7 @@ mod dirty_usage; mod guards; mod io_cache; mod io_cycle; +pub(crate) use io_cycle::nsscanner_with_storage_status; mod io_disk; #[cfg(test)] mod publish_gate_tests; diff --git a/crates/scanner/src/scanner_io/io_cycle.rs b/crates/scanner/src/scanner_io/io_cycle.rs index ce9236277..f51caf381 100644 --- a/crates/scanner/src/scanner_io/io_cycle.rs +++ b/crates/scanner/src/scanner_io/io_cycle.rs @@ -47,434 +47,432 @@ impl ScannerIOCycle for ECStore { leader_epoch: u64, scan_mode: HealScanMode, ) -> Result { - let child_token = ctx.child_token(); - let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch); + nsscanner_with_storage_status(self, ctx, budget, updates, want_cycle, leader_epoch, scan_mode).await + } +} - // Check the local pool metadata before listing buckets. A failed or - // canceled decommission remains suspended after its worker exits, so - // starting a scan in that state could build a snapshot that cannot be - // routed to the authoritative metadata object. - if self.scanner_data_usage_publication_blocked().await { +pub(crate) async fn nsscanner_with_storage_status( + store: &S, + ctx: CancellationToken, + budget: Arc, + updates: mpsc::Sender, + want_cycle: u64, + leader_epoch: u64, + scan_mode: HealScanMode, +) -> Result +where + S: ScannerStorage, +{ + let child_token = ctx.child_token(); + let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch); + + // Check the local pool metadata before listing buckets. A failed or + // canceled decommission remains suspended after its worker exits, so + // starting a scan in that state could build a snapshot that cannot be + // routed to the authoritative metadata object. + if store.scanner_data_usage_publication_blocked().await { + debug!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_SET_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + state = "cycle_data_usage_route_blocked", + "Scanner cycle deferred while data usage metadata remains hidden by data movement" + ); + return Ok(ScannerCycleResult::new( + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement), + None, + )); + } + + // Capture one storage-owned movement epoch for the entire cycle. Set + // workers must not each observe a fresh epoch: a movement transition + // between sets would otherwise allow a mixed-generation aggregate. + let publication_epoch = match store.scanner_data_usage_publication_admission().await { + Some(admission) => Some(admission.epoch()), + None => { + return Ok(ScannerCycleResult::new( + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement), + None, + )); + } + }; + + let distributed = store.setup_is_dist_erasure().await; + let activity_before = match scanner_activity_preflight(crate::scanner::probe_scanner_activity(store, distributed).await) { + ScannerActivityPreflight::Ready(snapshot) => snapshot, + ScannerActivityPreflight::ActivityBaselineUnavailable(err) => { + warn!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_SET_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + state = "cycle_activity_baseline_failed", + error = %err, + "Scanner cycle skipped because cluster activity could not be baselined" + ); + return Ok(ScannerCycleResult::new( + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable), + None, + )); + } + ScannerActivityPreflight::DataMovement => { debug!( target: "rustfs::scanner::io", event = EVENT_SCANNER_SET_STATE, component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_IO, - state = "cycle_data_usage_route_blocked", - "Scanner cycle deferred while data usage metadata remains hidden by data movement" + state = "cycle_data_movement_active", + "Scanner cycle deferred while rebalance or decommission data movement is active" ); return Ok(ScannerCycleResult::new( ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement), None, )); } - - // Capture one storage-owned movement epoch for the entire cycle. Set - // workers must not each observe a fresh epoch: a movement transition - // between sets would otherwise allow a mixed-generation aggregate. - let publication_epoch = match self.scanner_data_usage_publication_admission().await { - Some(admission) => Some(admission.epoch()), - None => { - return Ok(ScannerCycleResult::new( - ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement), - None, - )); - } - }; - - let distributed = self.setup_is_dist_erasure().await; - let activity_before = match scanner_activity_preflight(crate::scanner::probe_scanner_activity(self, distributed).await) { - ScannerActivityPreflight::Ready(snapshot) => snapshot, - ScannerActivityPreflight::ActivityBaselineUnavailable(err) => { - warn!( - target: "rustfs::scanner::io", - event = EVENT_SCANNER_SET_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_IO, - state = "cycle_activity_baseline_failed", - error = %err, - "Scanner cycle skipped because cluster activity could not be baselined" - ); - return Ok(ScannerCycleResult::new( - ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable), - None, - )); - } - ScannerActivityPreflight::DataMovement => { - debug!( - target: "rustfs::scanner::io", - event = EVENT_SCANNER_SET_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_IO, - state = "cycle_data_movement_active", - "Scanner cycle deferred while rebalance or decommission data movement is active" - ); - return Ok(ScannerCycleResult::new( - ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement), - None, - )); - } - }; - let dirty_generation_before_bucket_list = dirty_usage_generation(); - let bucket_listing = self.list_bucket_for_scanner(&BucketOptions::default()).await?; - let mut bucket_plan_complete = bucket_listing.topology_complete; - let all_buckets = Arc::new(bucket_listing.buckets); - let expected_sources = Arc::new( - self.pools - .iter() - .flat_map(|pool| { - pool.disk_set - .iter() - .map(|set| DataUsageCacheSource::new(set.pool_index, set.set_index)) - }) - .collect::>(), - ); - let mut buckets_by_source = HashMap::with_capacity(bucket_listing.set_buckets.len()); - for scope in bucket_listing.set_buckets { - let source = DataUsageCacheSource::new(scope.pool_index, scope.set_index); - if buckets_by_source.insert(source, scope.buckets).is_some() { - bucket_plan_complete = false; - } + }; + let dirty_generation_before_bucket_list = dirty_usage_generation(); + let bucket_listing = store.list_bucket_for_scanner(&BucketOptions::default()).await?; + let mut bucket_plan_complete = bucket_listing.topology_complete; + let all_buckets = Arc::new(bucket_listing.buckets); + let set_disks = store.all_set_disks(); + let expected_sources = Arc::new( + set_disks + .iter() + .map(|set| DataUsageCacheSource::new(set.pool_index, set.set_index)) + .collect::>(), + ); + let mut buckets_by_source = HashMap::with_capacity(bucket_listing.set_buckets.len()); + for scope in bucket_listing.set_buckets { + let source = DataUsageCacheSource::new(scope.pool_index, scope.set_index); + if buckets_by_source.insert(source, scope.buckets).is_some() { + bucket_plan_complete = false; } - bucket_plan_complete &= buckets_by_source.keys().copied().collect::>() == *expected_sources; - let scan_plan_digest = - scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_snapshot_digest(&activity_before)); - let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list)); - let cache_cycle_floor = Arc::new(AtomicU64::new(want_cycle)); - let tier_registry = runtime_tier_registry_for_cycle(want_cycle, leader_epoch).await; - let tier_registry_generation = tier_registry.generation; + } + bucket_plan_complete &= buckets_by_source.keys().copied().collect::>() == *expected_sources; + let scan_plan_digest = + scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_snapshot_digest(&activity_before)); + let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list)); + let cache_cycle_floor = Arc::new(AtomicU64::new(want_cycle)); + let tier_registry = runtime_tier_registry_for_cycle(want_cycle, leader_epoch).await; + let tier_registry_generation = tier_registry.generation; - if all_buckets.is_empty() { - reset_set_scan_gauges(); - if !bucket_plan_complete { - return Ok( - ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_publication_epoch(publication_epoch) - ); - } - let (activity_status, remote_publication_lease_targets) = - scanner_cycle_activity_status(self, distributed, &activity_before).await; - let dirty_usage_status = dirty_usage_snapshot_status(&dirty_usage_snapshot); - let status = classify_nsscanner_cycle( - true, - false, - ctx.is_cancelled(), - ScannerBucketScanStatus::Complete, - dirty_usage_status, - activity_status, - ); - let empty_usage = DataUsageInfo { - last_update: Some(SystemTime::now()), - scanner_cycle: Some(want_cycle), - usage_snapshot_complete: true, - ..Default::default() - }; - let observational_snapshot_published = if should_publish_observational_snapshot(status) { - publish_observational_snapshot(&updates, empty_usage).await? - } else { - publish_usage_snapshot(&updates, status, empty_usage).await? - }; - if !observational_snapshot_published { - return Ok(ScannerCycleResult::new(status, None).with_publication_epoch(publication_epoch)); - } - if status == ScannerCycleStatus::Complete { - complete_tier_registry_cycle(want_cycle, leader_epoch); - } - let dirty_usage_clear = - (status == ScannerCycleStatus::Complete).then(|| dirty_usage_snapshot.buckets.as_ref().clone()); - let remote_dirty_usage_acknowledgements = if status == ScannerCycleStatus::Complete { - crate::scanner::scanner_dirty_usage_acknowledgements(&activity_before) - } else { - Vec::new() - }; - return Ok(ScannerCycleResult::new(status, dirty_usage_clear) - .with_publication_epoch(publication_epoch) - .with_observational_snapshot_published(observational_snapshot_published) - .with_remote_publication_lease_targets(remote_publication_lease_targets) - .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)); - } - - let total_results = expected_sources.len(); - if total_results == 0 { - warn!( - target: "rustfs::scanner::io", - event = EVENT_SCANNER_SET_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_IO, - bucket_count = all_buckets.len(), - state = "no_disk_sets", - "Scanner set state update detected missing disk sets" - ); - reset_set_scan_gauges(); + if all_buckets.is_empty() { + reset_set_scan_gauges(); + if !bucket_plan_complete { 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 (activity_status, remote_publication_lease_targets) = + scanner_cycle_activity_status(store, distributed, &activity_before).await; + let dirty_usage_status = dirty_usage_snapshot_status(&dirty_usage_snapshot); + let status = classify_nsscanner_cycle( + true, + false, + ctx.is_cancelled(), + ScannerBucketScanStatus::Complete, + dirty_usage_status, + activity_status, ); - let bucket_failures = ScannerBucketFailureState::default(); - let pending_maintenance_work = Arc::new(AtomicBool::new(false)); - record_set_scan_concurrency_limit(set_scan_limit); - debug!( + let empty_usage = DataUsageInfo { + last_update: Some(SystemTime::now()), + scanner_cycle: Some(want_cycle), + usage_snapshot_complete: true, + ..Default::default() + }; + let observational_snapshot_published = if should_publish_observational_snapshot(status) { + publish_observational_snapshot(&updates, empty_usage).await? + } else { + publish_usage_snapshot(&updates, status, empty_usage).await? + }; + if !observational_snapshot_published { + return Ok(ScannerCycleResult::new(status, None).with_publication_epoch(publication_epoch)); + } + if status == ScannerCycleStatus::Complete { + complete_tier_registry_cycle(want_cycle, leader_epoch); + } + let dirty_usage_clear = (status == ScannerCycleStatus::Complete).then(|| dirty_usage_snapshot.buckets.as_ref().clone()); + let remote_dirty_usage_acknowledgements = if status == ScannerCycleStatus::Complete { + crate::scanner::scanner_dirty_usage_acknowledgements(&activity_before) + } else { + Vec::new() + }; + return Ok(ScannerCycleResult::new(status, dirty_usage_clear) + .with_publication_epoch(publication_epoch) + .with_observational_snapshot_published(observational_snapshot_published) + .with_remote_publication_lease_targets(remote_publication_lease_targets) + .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)); + } + + let total_results = expected_sources.len(); + if total_results == 0 { + warn!( 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", - "Scanner set concurrency budget resolved" + bucket_count = all_buckets.len(), + state = "no_disk_sets", + "Scanner set state update detected missing disk sets" ); - let set_scan_semaphore = Arc::new(Semaphore::new(set_scan_limit)); - let queued_set_scans = Arc::new(AtomicUsize::new(total_results)); - let active_set_scans = Arc::new(AtomicUsize::new(0)); - record_set_scans_queued(total_results); - record_set_scans_active(0); + reset_set_scan_gauges(); + return Ok(ScannerCycleResult::new(ScannerCycleStatus::Incomplete, None).with_publication_epoch(publication_epoch)); + } - let results = vec![DataUsageCache::default(); total_results]; - let results_mutex: Arc>> = Arc::new(Mutex::new(results)); - let first_err_mutex: Arc>> = Arc::new(Mutex::new(None)); - let mut results_index = 0usize; - let mut wait_futs = Vec::new(); + let set_scan_limit = scanner_budgeted_concurrency_limit( + scanner_max_concurrent_set_scans(total_results), + budget.requires_serial_progress_accounting(), + ); + let bucket_failures = ScannerBucketFailureState::default(); + let pending_maintenance_work = Arc::new(AtomicBool::new(false)); + record_set_scan_concurrency_limit(set_scan_limit); + debug!( + 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", + "Scanner set concurrency budget resolved" + ); + let set_scan_semaphore = Arc::new(Semaphore::new(set_scan_limit)); + let queued_set_scans = Arc::new(AtomicUsize::new(total_results)); + let active_set_scans = Arc::new(AtomicUsize::new(0)); + record_set_scans_queued(total_results); + record_set_scans_active(0); - for pool in self.pools.iter() { - for set in pool.disk_set.iter() { - let results_index_clone = results_index; - results_index += 1; - // Clone the Arc to move it into the spawned task - let set_clone: Arc = Arc::clone(set); - let source = DataUsageCacheSource::new(set.pool_index, set.set_index); - let set_buckets = buckets_by_source.remove(&source).unwrap_or_default(); - let pool_label = set.pool_index.to_string(); - let set_label = set.set_index.to_string(); + let results = vec![DataUsageCache::default(); total_results]; + let results_mutex: Arc>> = Arc::new(Mutex::new(results)); + let first_err_mutex: Arc>> = Arc::new(Mutex::new(None)); + let mut wait_futs = Vec::new(); - let child_token_clone = child_token.clone(); - let budget_clone = budget.clone(); - let want_cycle_clone = want_cycle; - let scan_mode_clone = scan_mode; - let results_mutex_clone = results_mutex.clone(); - let first_err_mutex_clone = first_err_mutex.clone(); - let set_scan_semaphore_clone = set_scan_semaphore.clone(); - let queued_set_scans_clone = queued_set_scans.clone(); - let active_set_scans_clone = active_set_scans.clone(); + for (results_index, set) in set_disks.iter().enumerate() { + let results_index_clone = results_index; + // Clone the Arc to move it into the spawned task + let set_clone: Arc = Arc::clone(set); + let source = DataUsageCacheSource::new(set.pool_index, set.set_index); + let set_buckets = buckets_by_source.remove(&source).unwrap_or_default(); + let pool_label = set.pool_index.to_string(); + let set_label = set.set_index.to_string(); - let (tx, mut rx) = mpsc::channel::(1); - let failed_scope_tx = tx.clone(); + let child_token_clone = child_token.clone(); + let budget_clone = budget.clone(); + let want_cycle_clone = want_cycle; + let scan_mode_clone = scan_mode; + let results_mutex_clone = results_mutex.clone(); + let first_err_mutex_clone = first_err_mutex.clone(); + let set_scan_semaphore_clone = set_scan_semaphore.clone(); + let queued_set_scans_clone = queued_set_scans.clone(); + let active_set_scans_clone = active_set_scans.clone(); - // Spawn task to receive and store results - let receiver_fut = tokio::spawn(async move { - while let Some(result) = rx.recv().await { - let mut results = results_mutex_clone.lock().await; - results[results_index_clone] = result; - } - }); - wait_futs.push(AbortOnDropHandle::new(receiver_fut)); + let (tx, mut rx) = mpsc::channel::(1); + let failed_scope_tx = tx.clone(); - let scan_plan = ScannerBucketScanPlan { - buckets: set_buckets, - all_buckets: Arc::clone(&all_buckets), - digest: scan_plan_digest, - leader_epoch, - tier_registry_generation, - publication_epoch, - dirty_usage_buckets: dirty_usage_snapshot.buckets.clone(), - bucket_failures: bucket_failures.clone(), - pending_maintenance_work: pending_maintenance_work.clone(), - cache_cycle_floor: cache_cycle_floor.clone(), - }; - // Spawn task to run the scanner - let scanner_fut = tokio::spawn(async move { - let permit_wait = child_token_clone.clone(); - let permit_wait_start = Instant::now(); - let _permit = tokio::select! { - permit = set_scan_semaphore_clone.acquire_owned() => match permit { - Ok(permit) => permit, - Err(_) => return, - }, - _ = permit_wait.cancelled() => return, - }; - metrics::histogram!( - METRIC_SCANNER_SET_SCAN_WAIT_SECONDS, - "pool" => pool_label.clone(), - "set" => set_label.clone() - ) - .record(permit_wait_start.elapsed().as_secs_f64()); - let queued_count = decrement_atomic_usize(&queued_set_scans_clone); - record_set_scans_queued(queued_count); - let _active_guard = SetScanActiveGuard::new(active_set_scans_clone); - - if let Err(e) = set_clone - .nsscanner_cache( - child_token_clone.clone(), - budget_clone, - scan_plan, - tx, - want_cycle_clone, - scan_mode_clone, - ) - .await - { - if child_token_clone.is_cancelled() { - debug!( - pool = %pool_label, - set = %set_label, - error = %e, - "Scanner set scan stopped after cancellation" - ); - return; - } - - counter!( - "rustfs_scanner_set_failure_total", - "pool" => pool_label.clone(), - "set" => set_label.clone(), - "stage" => "nsscanner_cache".to_string() - ) - .increment(1); - error!( - target: "rustfs::scanner::io", - event = EVENT_SCANNER_SET_STATE, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_IO, - pool = %pool_label, - set = %set_label, - error = %e, - state = "set_scan_failed", - "Scanner set scan failed; continuing cycle" - ); - let _ = failed_scope_tx - .send(DataUsageCache { - info: DataUsageCacheInfo { - name: DATA_USAGE_ROOT.to_string(), - next_cycle: want_cycle_clone, - leader_epoch, - source: Some(source), - snapshot_complete: false, - scan_plan_digest: Some(scan_plan_digest), - cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, - ..Default::default() - }, - cache: HashMap::new(), - }) - .await; - let mut first_err = first_err_mutex_clone.lock().await; - record_set_scan_failure(&mut first_err, e); - } - }); - wait_futs.push(AbortOnDropHandle::new(scanner_fut)); + // Spawn task to receive and store results + let receiver_fut = tokio::spawn(async move { + while let Some(result) = rx.recv().await { + let mut results = results_mutex_clone.lock().await; + results[results_index_clone] = result; } - } + }); + wait_futs.push(AbortOnDropHandle::new(receiver_fut)); - for join_result in join_all(wait_futs).await { - if let Err(err) = join_result { + let scan_plan = ScannerBucketScanPlan { + buckets: set_buckets, + all_buckets: Arc::clone(&all_buckets), + digest: scan_plan_digest, + leader_epoch, + tier_registry_generation, + publication_epoch, + dirty_usage_buckets: dirty_usage_snapshot.buckets.clone(), + bucket_failures: bucket_failures.clone(), + pending_maintenance_work: pending_maintenance_work.clone(), + cache_cycle_floor: cache_cycle_floor.clone(), + }; + // Spawn task to run the scanner + let scanner_fut = tokio::spawn(async move { + let permit_wait = child_token_clone.clone(); + let permit_wait_start = Instant::now(); + let _permit = tokio::select! { + permit = set_scan_semaphore_clone.acquire_owned() => match permit { + Ok(permit) => permit, + Err(_) => return, + }, + _ = permit_wait.cancelled() => return, + }; + metrics::histogram!( + METRIC_SCANNER_SET_SCAN_WAIT_SECONDS, + "pool" => pool_label.clone(), + "set" => set_label.clone() + ) + .record(permit_wait_start.elapsed().as_secs_f64()); + let queued_count = decrement_atomic_usize(&queued_set_scans_clone); + record_set_scans_queued(queued_count); + let _active_guard = SetScanActiveGuard::new(active_set_scans_clone); + + if let Err(e) = set_clone + .nsscanner_cache(child_token_clone.clone(), budget_clone, scan_plan, tx, want_cycle_clone, scan_mode_clone) + .await + { + if child_token_clone.is_cancelled() { + debug!( + pool = %pool_label, + set = %set_label, + error = %e, + "Scanner set scan stopped after cancellation" + ); + return; + } + + counter!( + "rustfs_scanner_set_failure_total", + "pool" => pool_label.clone(), + "set" => set_label.clone(), + "stage" => "nsscanner_cache".to_string() + ) + .increment(1); error!( target: "rustfs::scanner::io", event = EVENT_SCANNER_SET_STATE, component = LOG_COMPONENT_SCANNER, subsystem = LOG_SUBSYSTEM_IO, - state = "set_task_join_failed", - error = %err, - "Scanner set task join failed" + pool = %pool_label, + set = %set_label, + error = %e, + state = "set_scan_failed", + "Scanner set scan failed; continuing cycle" ); - let mut first_err = first_err_mutex.lock().await; - record_set_scan_failure(&mut first_err, scanner_task_join_error("scanner set", err)); + let _ = failed_scope_tx + .send(DataUsageCache { + info: DataUsageCacheInfo { + name: DATA_USAGE_ROOT.to_string(), + next_cycle: want_cycle_clone, + leader_epoch, + source: Some(source), + snapshot_complete: false, + scan_plan_digest: Some(scan_plan_digest), + cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, + ..Default::default() + }, + cache: HashMap::new(), + }) + .await; + let mut first_err = first_err_mutex_clone.lock().await; + record_set_scan_failure(&mut first_err, e); } - } - record_set_scan_concurrency_limit(0); - record_set_scans_queued(0); - record_set_scans_active(0); + }); + wait_futs.push(AbortOnDropHandle::new(scanner_fut)); + } - let first_err = first_err_mutex.lock().await.take(); - let results = results_mutex.lock().await.clone(); - let completed_all_sets = bucket_plan_complete && scanner_results_form_complete_snapshot(&results, &expected_sources); - let result = finalize_nsscanner_result(&results, first_err); - let failed_buckets = bucket_failures.hard.lock().await.clone(); - let partial_buckets = bucket_failures.partial.lock().await.clone(); - let namespace_not_found_buckets = bucket_failures.namespace_not_found.lock().await.clone(); - let scan_scope_matches = scanner_results_match_scan_scope(&results, &expected_sources); - let bucket_scan_status = scanner_bucket_scan_status( - !failed_buckets.is_empty(), - scan_scope_matches && !partial_buckets.is_empty(), - scan_scope_matches && !namespace_not_found_buckets.is_empty(), - ); - let pending_maintenance_work = pending_maintenance_work_for_cycle(&pending_maintenance_work, &results); - let observed_cycle_floor = cache_cycle_floor.load(Ordering::Acquire); - let required_cycle_floor = (observed_cycle_floor > want_cycle).then_some(observed_cycle_floor); - let budget_elapsed = budget.budget_elapsed(); - let dirty_usage_status = dirty_usage_snapshot_status(&dirty_usage_snapshot); - let dirty_usage_current = dirty_usage_status == DirtyUsageSnapshotStatus::Current; - let (activity_status, remote_publication_lease_targets) = - scanner_cycle_activity_status(self, distributed, &activity_before).await; - let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::>(); - let completed_usage = completed_data_usage_info( - &results, - &expected_sources, - &all_bucket_names, - &tier_registry.names, - bucket_plan_complete, - budget_elapsed, - ctx.is_cancelled(), - ); - let observational_usage = completed_usage - .is_none() - .then(|| { - observational_data_usage_info( - &results, - &expected_sources, - &all_bucket_names, - &tier_registry.names, - scan_plan_digest, - want_cycle, - leader_epoch, - ) - }) - .flatten(); - let structurally_complete_snapshot = result.is_ok() && completed_all_sets && completed_usage.is_some(); - let cycle_status = classify_nsscanner_cycle( - structurally_complete_snapshot, - budget_elapsed, - ctx.is_cancelled(), - bucket_scan_status, - dirty_usage_status, - activity_status, - ); - let observational_snapshot_published = if let Some((data_usage_info, _)) = completed_usage { - if should_publish_observational_snapshot(cycle_status) { - publish_observational_snapshot(&updates, data_usage_info).await? - } else { - publish_usage_snapshot(&updates, cycle_status, data_usage_info).await? - } - } else if !ctx.is_cancelled() - && let Some((data_usage_info, _)) = observational_usage - { + for join_result in join_all(wait_futs).await { + if let Err(err) = join_result { + error!( + target: "rustfs::scanner::io", + event = EVENT_SCANNER_SET_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_IO, + state = "set_task_join_failed", + error = %err, + "Scanner set task join failed" + ); + let mut first_err = first_err_mutex.lock().await; + record_set_scan_failure(&mut first_err, scanner_task_join_error("scanner set", err)); + } + } + record_set_scan_concurrency_limit(0); + record_set_scans_queued(0); + record_set_scans_active(0); + + let first_err = first_err_mutex.lock().await.take(); + let results = results_mutex.lock().await.clone(); + let completed_all_sets = bucket_plan_complete && scanner_results_form_complete_snapshot(&results, &expected_sources); + let result = finalize_nsscanner_result(&results, first_err); + let failed_buckets = bucket_failures.hard.lock().await.clone(); + let partial_buckets = bucket_failures.partial.lock().await.clone(); + let namespace_not_found_buckets = bucket_failures.namespace_not_found.lock().await.clone(); + let scan_scope_matches = scanner_results_match_scan_scope(&results, &expected_sources); + let bucket_scan_status = scanner_bucket_scan_status( + !failed_buckets.is_empty(), + scan_scope_matches && !partial_buckets.is_empty(), + scan_scope_matches && !namespace_not_found_buckets.is_empty(), + ); + let pending_maintenance_work = pending_maintenance_work_for_cycle(&pending_maintenance_work, &results); + let observed_cycle_floor = cache_cycle_floor.load(Ordering::Acquire); + let required_cycle_floor = (observed_cycle_floor > want_cycle).then_some(observed_cycle_floor); + let budget_elapsed = budget.budget_elapsed(); + let dirty_usage_status = dirty_usage_snapshot_status(&dirty_usage_snapshot); + let dirty_usage_current = dirty_usage_status == DirtyUsageSnapshotStatus::Current; + let (activity_status, remote_publication_lease_targets) = + scanner_cycle_activity_status(store, distributed, &activity_before).await; + let all_bucket_names = all_buckets.iter().map(|bucket| bucket.name.clone()).collect::>(); + let completed_usage = completed_data_usage_info( + &results, + &expected_sources, + &all_bucket_names, + &tier_registry.names, + bucket_plan_complete, + budget_elapsed, + ctx.is_cancelled(), + ); + let observational_usage = completed_usage + .is_none() + .then(|| { + observational_data_usage_info( + &results, + &expected_sources, + &all_bucket_names, + &tier_registry.names, + scan_plan_digest, + want_cycle, + leader_epoch, + ) + }) + .flatten(); + let structurally_complete_snapshot = result.is_ok() && completed_all_sets && completed_usage.is_some(); + let cycle_status = classify_nsscanner_cycle( + structurally_complete_snapshot, + budget_elapsed, + ctx.is_cancelled(), + bucket_scan_status, + dirty_usage_status, + activity_status, + ); + let observational_snapshot_published = if let Some((data_usage_info, _)) = completed_usage { + if should_publish_observational_snapshot(cycle_status) { publish_observational_snapshot(&updates, data_usage_info).await? } else { - false - }; - let dirty_usage_clear = should_clear_dirty_usage_snapshot( - result.is_ok(), - structurally_complete_snapshot, - budget_elapsed, - activity_status == ScannerCycleActivityStatus::Unchanged && dirty_usage_current, - &dirty_usage_snapshot.buckets, - &failed_buckets, - ); - result?; - if cycle_status == ScannerCycleStatus::Complete { - complete_tier_registry_cycle(want_cycle, leader_epoch); + publish_usage_snapshot(&updates, cycle_status, data_usage_info).await? } - let remote_dirty_usage_acknowledgements = if cycle_status == ScannerCycleStatus::Complete { - crate::scanner::scanner_dirty_usage_acknowledgements(&activity_before) - } else { - Vec::new() - }; - Ok(ScannerCycleResult::new(cycle_status, dirty_usage_clear) - .with_publication_epoch(publication_epoch) - .with_observational_snapshot_published(observational_snapshot_published) - .with_remote_publication_lease_targets(remote_publication_lease_targets) - .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements) - .with_failed_dirty_usage(!failed_buckets.is_empty()) - .with_pending_maintenance_work(pending_maintenance_work) - .with_required_cycle_floor(required_cycle_floor)) + } else if !ctx.is_cancelled() + && let Some((data_usage_info, _)) = observational_usage + { + publish_observational_snapshot(&updates, data_usage_info).await? + } else { + false + }; + let dirty_usage_clear = should_clear_dirty_usage_snapshot( + result.is_ok(), + structurally_complete_snapshot, + budget_elapsed, + activity_status == ScannerCycleActivityStatus::Unchanged && dirty_usage_current, + &dirty_usage_snapshot.buckets, + &failed_buckets, + ); + result?; + if cycle_status == ScannerCycleStatus::Complete { + complete_tier_registry_cycle(want_cycle, leader_epoch); } + let remote_dirty_usage_acknowledgements = if cycle_status == ScannerCycleStatus::Complete { + crate::scanner::scanner_dirty_usage_acknowledgements(&activity_before) + } else { + Vec::new() + }; + Ok(ScannerCycleResult::new(cycle_status, dirty_usage_clear) + .with_publication_epoch(publication_epoch) + .with_observational_snapshot_published(observational_snapshot_published) + .with_remote_publication_lease_targets(remote_publication_lease_targets) + .with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements) + .with_failed_dirty_usage(!failed_buckets.is_empty()) + .with_pending_maintenance_work(pending_maintenance_work) + .with_required_cycle_floor(required_cycle_floor)) } diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index 7df444cbd..9efd990be 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -15,7 +15,11 @@ use std::collections::HashMap; use std::sync::Arc; +#[cfg(test)] +use http::HeaderMap; +use rustfs_lock::NamespaceLockWrapper; use serde::{Deserialize, Serialize}; +use tokio::sync::Notify; pub(crate) use s3s::dto::{ BucketLifecycleConfiguration as EcstoreBucketLifecycleConfiguration, LifecycleRuleFilter as EcstoreLifecycleRuleFilter, @@ -29,11 +33,7 @@ pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_audit::L pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::{ apply_expiry_rule as ecstore_apply_expiry_rule, apply_transition_rule as ecstore_apply_transition_rule, }; -pub(crate) use rustfs_ecstore::api::bucket::lifecycle::evaluator::Evaluator as EcstoreEvaluator; -pub(crate) use rustfs_ecstore::api::bucket::lifecycle::lifecycle::{ - Event as EcstoreEvent, Lifecycle as EcstoreLifecycle, ObjectOpts as EcstoreObjectOpts, - TRANSITION_COMPLETE as ECSTORE_TRANSITION_COMPLETE, object_opts_from_object_info as ecstore_object_opts_from_object_info, -}; +pub(crate) use rustfs_ecstore::api::bucket::lifecycle::lifecycle::object_opts_from_object_info as ecstore_object_opts_from_object_info; #[cfg(test)] pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::init_bucket_metadata_sys as ecstore_init_bucket_metadata_sys; pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::{ @@ -92,6 +92,7 @@ pub(crate) use rustfs_ecstore::api::event::{EventArgs as EcstoreEventArgs, send_ pub(crate) use rustfs_ecstore::api::layout::{ EndpointServerPools as EcstoreEndpointServerPools, Endpoints as EcstoreEndpoints, PoolEndpoints as EcstorePoolEndpoints, }; +pub(crate) use rustfs_ecstore::api::notification::NotificationSys as EcstoreNotificationSys; pub(crate) use rustfs_ecstore::api::notification::scanner_peer_transport_error_message_is_retryable; pub(crate) use rustfs_ecstore::api::object::{ SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitState, @@ -101,19 +102,25 @@ pub(crate) use rustfs_ecstore::api::rebalance::{ RebalStatus as EcstoreRebalStatus, RebalanceInfo as EcstoreRebalanceInfo, RebalanceMeta as EcstoreRebalanceMeta, RebalanceStats as EcstoreRebalanceStats, }; +pub(crate) use rustfs_ecstore::api::rpc::ScannerBucketListing as EcstoreScannerBucketListing; #[cfg(test)] pub(crate) use rustfs_ecstore::api::runtime::InstanceContext as EcstoreInstanceContext; pub(crate) use rustfs_ecstore::api::runtime::{ expiry_state_handle as ecstore_expiry_state_handle, global_tier_config_mgr as ecstore_get_global_tier_config_mgr, object_store_handle as ecstore_resolve_object_store_handle, setup_is_erasure as ecstore_is_erasure, - setup_is_erasure_sd as ecstore_is_erasure_sd, }; pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks; -pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore; #[cfg(test)] pub(crate) use rustfs_ecstore::api::storage::SCANNER_PUBLICATION_LEASE_TTL_MS as ECSTORE_SCANNER_PUBLICATION_LEASE_TTL_MS; #[cfg(test)] pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx; +pub(crate) use rustfs_ecstore::api::storage::{ + ECStore as EcstoreStore, ScannerDataMovementPauseStatus as EcstoreScannerDataMovementPauseStatus, +}; +pub(crate) use rustfs_lifecycle::{ + Evaluator as EcstoreEvaluator, Event as EcstoreEvent, Lifecycle as EcstoreLifecycle, ObjectOpts as EcstoreObjectOpts, + TRANSITION_COMPLETE as ECSTORE_TRANSITION_COMPLETE, +}; use rustfs_storage_api as storage_contracts; pub(crate) mod owner { @@ -134,11 +141,11 @@ pub(crate) mod owner { ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache, - ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_erasure_sd, - ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info, - ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_read_config, - ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config, - ecstore_send_event, scanner_replication_config_for_lifecycle_eval, + ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_reserved_or_invalid_bucket, + ecstore_list_path_raw, ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, + ecstore_path2_bucket_object_with_base_path, ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, + ecstore_resolve_object_store_handle, ecstore_save_config, ecstore_send_event, + scanner_replication_config_for_lifecycle_eval, }; #[cfg(test)] @@ -289,9 +296,10 @@ impl From for ScannerReplicationHealResult { } pub(crate) mod scan { + #[cfg(test)] + pub(crate) use super::storage_contracts::BucketOperations; pub(crate) use super::storage_contracts::{ - BucketOperations, BucketOptions, NamespaceLocking, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, - SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, + BucketOptions, NamespaceLocking, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, }; #[cfg(test)] pub(crate) use super::storage_contracts::{DeleteBucketOptions, MakeBucketOptions, ObjectIO}; @@ -303,3 +311,263 @@ pub(crate) mod scanner_io { #[cfg(test)] pub(crate) use super::storage_contracts::{HTTPRangeSpec, ObjectIO}; } + +pub(crate) type ScannerBucketListing = EcstoreScannerBucketListing; +pub(crate) type ScannerDataMovementPauseStatus = EcstoreScannerDataMovementPauseStatus; +pub(crate) type ScannerNotificationSys = EcstoreNotificationSys; + +#[async_trait::async_trait] +pub(crate) trait ScannerStorage: + crate::ScannerObjectIO + + crate::ScannerConfigObjectDelete + + storage_contracts::BucketOperations + + storage_contracts::NamespaceLocking +{ + fn scanner_topology_digest(&self) -> [u8; 32]; + fn scanner_namespace_mutation_generation(&self) -> u64; + async fn scanner_data_movement_activity(&self) -> (bool, bool, u64); + async fn scanner_data_usage_publication_blocked(&self) -> bool; + async fn scanner_data_movement_pause_status(&self) -> ScannerDataMovementPauseStatus; + fn scanner_data_movement_generation(&self) -> u64; + fn scanner_data_movement_changed(&self) -> Arc; + fn scanner_notification_system(&self) -> Option>; + async fn setup_is_erasure(&self) -> bool; + async fn setup_is_dist_erasure(&self) -> bool; + async fn setup_is_erasure_sd(&self) -> bool; + async fn list_bucket_for_scanner(&self, opts: &storage_contracts::BucketOptions) -> EcstoreResultType; + fn all_set_disks(&self) -> Vec>; + async fn scanner_pause_backlog_writable_set_disks(&self) -> Vec>; + #[cfg(test)] + fn scanner_observed_probe_store_key(&self) -> usize; +} + +#[async_trait::async_trait] +impl ScannerStorage for EcstoreStore { + fn scanner_topology_digest(&self) -> [u8; 32] { + crate::scanner::scanner_topology_digest(self) + } + + fn scanner_namespace_mutation_generation(&self) -> u64 { + EcstoreStore::scanner_namespace_mutation_generation(self) + } + + async fn scanner_data_movement_activity(&self) -> (bool, bool, u64) { + EcstoreStore::scanner_data_movement_activity(self).await + } + + async fn scanner_data_usage_publication_blocked(&self) -> bool { + EcstoreStore::scanner_data_usage_publication_blocked(self).await + } + + async fn scanner_data_movement_pause_status(&self) -> ScannerDataMovementPauseStatus { + EcstoreStore::scanner_data_movement_pause_status(self).await + } + + fn scanner_data_movement_generation(&self) -> u64 { + EcstoreStore::scanner_data_movement_generation(self) + } + + fn scanner_data_movement_changed(&self) -> Arc { + EcstoreStore::scanner_data_movement_changed(self) + } + + fn scanner_notification_system(&self) -> Option> { + EcstoreStore::notification_system(self) + } + + async fn setup_is_erasure(&self) -> bool { + EcstoreStore::setup_is_erasure(self).await + } + + async fn setup_is_dist_erasure(&self) -> bool { + EcstoreStore::setup_is_dist_erasure(self).await + } + + async fn setup_is_erasure_sd(&self) -> bool { + EcstoreStore::setup_is_erasure_sd(self).await + } + + async fn list_bucket_for_scanner(&self, opts: &storage_contracts::BucketOptions) -> EcstoreResultType { + EcstoreStore::list_bucket_for_scanner(self, opts).await + } + + fn all_set_disks(&self) -> Vec> { + EcstoreStore::all_set_disks(self) + } + + async fn scanner_pause_backlog_writable_set_disks(&self) -> Vec> { + EcstoreStore::scanner_pause_backlog_writable_set_disks(self).await + } + + #[cfg(test)] + fn scanner_observed_probe_store_key(&self) -> usize { + std::ptr::from_ref(self).cast::<()>() as usize + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Debug, Default)] + struct FakeScannerStorage; + + #[async_trait::async_trait] + impl storage_contracts::ObjectIO for FakeScannerStorage { + type Error = EcstoreErrorType; + type RangeSpec = storage_contracts::HTTPRangeSpec; + type HeaderMap = HeaderMap; + type ObjectOptions = ::ObjectOptions; + type ObjectInfo = ::ObjectInfo; + type GetObjectReader = ::GetObjectReader; + type PutObjectReader = ::PutObjectReader; + + async fn get_object_reader( + &self, + _bucket: &str, + _object: &str, + _range: Option, + _h: Self::HeaderMap, + _opts: &Self::ObjectOptions, + ) -> Result { + Err(EcstoreErrorType::other("fake scanner storage has no object reader")) + } + + async fn put_object( + &self, + _bucket: &str, + _object: &str, + _data: &mut Self::PutObjectReader, + _opts: &Self::ObjectOptions, + ) -> Result { + Err(EcstoreErrorType::other("fake scanner storage has no object writer")) + } + } + + #[async_trait::async_trait] + impl storage_contracts::BucketOperations for FakeScannerStorage { + type Error = EcstoreErrorType; + + async fn make_bucket(&self, _bucket: &str, _opts: &storage_contracts::MakeBucketOptions) -> Result<(), Self::Error> { + Err(EcstoreErrorType::other("fake scanner storage cannot make buckets")) + } + + async fn get_bucket_info( + &self, + _bucket: &str, + _opts: &storage_contracts::BucketOptions, + ) -> Result { + Err(EcstoreErrorType::other("fake scanner storage cannot read buckets")) + } + + async fn list_bucket( + &self, + _opts: &storage_contracts::BucketOptions, + ) -> Result, Self::Error> { + Ok(Vec::new()) + } + + async fn delete_bucket(&self, _bucket: &str, _opts: &storage_contracts::DeleteBucketOptions) -> Result<(), Self::Error> { + Err(EcstoreErrorType::other("fake scanner storage cannot delete buckets")) + } + } + + #[async_trait::async_trait] + impl storage_contracts::NamespaceLocking for FakeScannerStorage { + type Error = EcstoreErrorType; + type NamespaceLock = NamespaceLockWrapper; + + async fn new_ns_lock(&self, _bucket: &str, _object: &str) -> Result { + Err(EcstoreErrorType::other("fake scanner storage has no namespace lock")) + } + } + + #[async_trait::async_trait] + impl crate::ScannerConfigObjectDelete for FakeScannerStorage { + async fn delete_config_object( + &self, + _bucket: &str, + _object: &str, + _opts: ::ObjectOptions, + ) -> EcstoreResultType<::ObjectInfo> { + Err(EcstoreErrorType::other("fake scanner storage cannot delete objects")) + } + + async fn scanner_data_usage_publication_admission(&self) -> Option { + Some(crate::ScannerDataUsagePublicationAdmission::unfenced()) + } + } + + #[async_trait::async_trait] + impl ScannerStorage for FakeScannerStorage { + fn scanner_topology_digest(&self) -> [u8; 32] { + [0; 32] + } + + fn scanner_namespace_mutation_generation(&self) -> u64 { + 0 + } + + async fn scanner_data_movement_activity(&self) -> (bool, bool, u64) { + (false, false, 0) + } + + async fn scanner_data_usage_publication_blocked(&self) -> bool { + false + } + + async fn scanner_data_movement_pause_status(&self) -> ScannerDataMovementPauseStatus { + ScannerDataMovementPauseStatus::default() + } + + fn scanner_data_movement_generation(&self) -> u64 { + 0 + } + + fn scanner_data_movement_changed(&self) -> Arc { + Arc::new(Notify::new()) + } + + fn scanner_notification_system(&self) -> Option> { + None + } + + async fn setup_is_erasure(&self) -> bool { + false + } + + async fn setup_is_dist_erasure(&self) -> bool { + false + } + + async fn setup_is_erasure_sd(&self) -> bool { + true + } + + async fn list_bucket_for_scanner( + &self, + _opts: &storage_contracts::BucketOptions, + ) -> EcstoreResultType { + Err(EcstoreErrorType::other("fake scanner storage cannot list scanner buckets")) + } + + fn all_set_disks(&self) -> Vec> { + Vec::new() + } + + async fn scanner_pause_backlog_writable_set_disks(&self) -> Vec> { + Vec::new() + } + + fn scanner_observed_probe_store_key(&self) -> usize { + 0 + } + } + + fn assert_scanner_storage() {} + + #[test] + fn scanner_storage_contract_accepts_fake_without_ecstore() { + assert_scanner_storage::(); + } +} diff --git a/docs/architecture/crate-boundaries.md b/docs/architecture/crate-boundaries.md index e4b188a3e..fb94ed0aa 100644 --- a/docs/architecture/crate-boundaries.md +++ b/docs/architecture/crate-boundaries.md @@ -57,9 +57,9 @@ Heal is split by responsibility, not by the shared word "heal". ECStore owns era `crates/scanner` owns discovery, data-usage publication, lifecycle/replication scan actions, bitrot scan dispatch, and scanner-driven repair requests. Scanner may request repair through the heal channel, but it must not directly execute erasure-set repair primitives. -`rustfs-scanner-metrics` owns scanner telemetry DTOs, global scanner counters, lifecycle action labels consumed by metrics, and the short-window latency accumulator used by those metrics. ECStore, lifecycle, observability, admin, and scanner code may depend on this crate for metrics only. `rustfs-scanner-contracts` must not regain metrics, globals, or telemetry implementation; it is reserved for scanner storage or wire contract types. +`rustfs-scanner-metrics` owns scanner telemetry DTOs, global scanner counters, lifecycle action labels consumed by metrics, and the short-window latency accumulator used by those metrics. ECStore, lifecycle, observability, admin, and scanner code may depend on this crate for metrics only. Scanner storage seams currently live in `rustfs-scanner`'s `storage_api.rs`; if a future `rustfs-scanner-contracts` crate is reintroduced for shared storage or wire contracts, it must not regain metrics, globals, or telemetry implementation. -`remote_scanner` remains scanner-owned for now because it carries the scanner cycle fence, replay protection, stream envelope, and per-bucket scan result protocol. A future scanner storage seam may either move remote disk scan execution behind an ECStore storage capability or move the whole remote scanner protocol with scanner; leaving the wire protocol split across both sides without a documented owner is not allowed. +`remote_scanner` remains scanner-owned for #2219 because it carries the scanner cycle fence, replay protection, stream envelope, and per-bucket scan result protocol. The scanner storage seam exposes the store, set, and disk capabilities needed by the remote execution path, while the wire protocol stays physically owned by scanner. A future split may move remote disk scan execution behind an ECStore storage capability or move the whole remote scanner protocol with scanner; leaving the envelope, fence, replay cache, and execution path split across both sides without a documented owner is not allowed. The scanner usage authority decision is fixed in [scanner-usage-authority-decision.md](scanner-usage-authority-decision.md): scanner usage remains hard-quota authority. A future scanner storage seam must therefore model the concrete publication, cycle-lock, usage-floor, observed-snapshot, and recovery-marker capabilities described in [scanner-usage-publication.md](scanner-usage-publication.md), not a generic key-value abstraction.