gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-05 19:55:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(scanner): add storage seam (#7105)

Browse Source
This commit is contained in:
houseme
2026-09-03 20:31:59 +08:00
committed by GitHub
parent 53cabe9274
commit 62e66baf89
16 changed files with 850 additions and 560 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ARCHITECTURE.md
+1 -1
View File
@@ -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. |
Cargo.lock
Generated
+1 -4
View File
@@ -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"
Cargo.toml
-2
View File
@@ -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" }
crates/scanner-contracts/Cargo.toml
-35
View File
@@ -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
crates/scanner-contracts/src/lib.rs
-13
View File
@@ -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.
crates/scanner/Cargo.toml
+1
View File
@@ -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 }
crates/scanner/src/lib.rs
+9 -11
View File
@@ -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<ScannerDataUsagePublicationAdmission> {
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<Uuid>,
) -> Option<ScannerPublicationCommitScope> {
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<Uuid>,
lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
) -> Option<ScannerPublicationCommitScope> {
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<ScannerDataUsagePublicationAdmission> {
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<Uuid>,
) -> Option<ScannerPublicationCommitScope> {
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<Uuid>,
lease_release_safe: Arc<std::sync::atomic::AtomicBool>,
) -> Option<ScannerPublicationCommitScope> {
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,
crates/scanner/src/scanner.rs
+97 -53
View File
@@ -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<ECStore>) -> usize {
Arc::as_ptr(storeapi).cast::<()>() as usize
fn scanner_observed_probe_store_key<S>(storeapi: &Arc<S>) -> usize
where
S: ScannerStorage,
{
storeapi.scanner_observed_probe_store_key()
}
#[cfg(test)]
fn notify_scanner_runtime_observed_for_test(storeapi: &Arc<ECStore>, observation: ScannerPauseBacklogObservation) {
fn notify_scanner_runtime_observed_for_test<S>(storeapi: &Arc<S>, 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<ECStore>) -> bool {
async fn persisted_usage_cache_is_cold_for_startup<S>(storeapi: &Arc<S>) -> 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<ECStore>) -> b
}
}
async fn initial_scanner_startup_usage_state(storeapi: &Arc<ECStore>) -> (bool, bool) {
async fn initial_scanner_startup_usage_state<S>(storeapi: &Arc<S>) -> (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<ECStore>) {
init_data_scanner_with_storage(ctx, storeapi).await;
}
async fn init_data_scanner_with_storage<S>(ctx: CancellationToken, storeapi: Arc<S>)
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<ECStore>) -> ScannerMaintenanceFeatures {
async fn detect_scanner_maintenance_features<S>(storeapi: &Arc<S>) -> 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<ECStore>) -> Scanner
async fn detect_stable_scanner_maintenance_features(
ctx: &CancellationToken,
storeapi: &Arc<ECStore>,
storeapi: &Arc<impl ScannerStorage>,
) -> Option<(ScannerMaintenanceFeatures, u64)> {
detect_stable_scanner_maintenance_features_with(
ctx,
@@ -1259,7 +1281,7 @@ where
async fn configure_scanner_defaults(
ctx: &CancellationToken,
storeapi: &Arc<ECStore>,
storeapi: &Arc<impl ScannerStorage>,
) -> (ScannerMaintenanceFeatures, Option<u64>) {
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<S>(
ctx: &CancellationToken,
storeapi: &Arc<ECStore>,
storeapi: &Arc<S>,
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<S>(
ctx: &CancellationToken,
storeapi: &Arc<ECStore>,
storeapi: &Arc<S>,
cycle_info: &mut CurrentCycle,
cycle_revision: &mut DataUsageCacheRevision,
leader_epoch: u64,
cycle_budget: Arc<ScannerCycleBudget>,
) -> 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::<DataUsageInfo>(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<ECStore>) -> Result<(), ScannerError> {
run_data_scanner_with_storage(ctx, storeapi).await
}
async fn run_data_scanner_with_storage<S>(ctx: CancellationToken, storeapi: Arc<S>) -> 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<ECStore>) -> ScannerPauseBacklogObservation {
async fn current_scanner_pause_backlog_observation<S>(storeapi: &Arc<S>) -> 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<ECStore>) -> S
}
}
async fn wait_for_scanner_data_movement_resume(
async fn wait_for_scanner_data_movement_resume<S>(
ctx: &CancellationToken,
storeapi: &Arc<ECStore>,
storeapi: &Arc<S>,
guard: &NamespaceLockGuard,
pause_backlog: &mut ScannerPauseBacklogController,
) -> bool {
pause_backlog: &mut ScannerPauseBacklogController<S>,
) -> 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<ECStore>,
async fn finish_scanner_pause_backlog_cycle<S>(
pause_backlog: &mut ScannerPauseBacklogController<S>,
storeapi: &Arc<S>,
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<S>(
ctx: CancellationToken,
storeapi: Arc<ECStore>,
storeapi: Arc<S>,
mut maintenance_features: ScannerMaintenanceFeatures,
mut maintenance_generation_seen: Option<u64>,
) -> 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<ScannerCycleDeferReason> {
async fn final_data_usage_publication_defer_reason<S>(storeapi: &S, status: ScannerCycleStatus) -> Option<ScannerCycleDeferReason>
where
S: ScannerStorage,
{
match status {
ScannerCycleStatus::Complete | ScannerCycleStatus::Superseded => {
if storeapi.scanner_data_usage_publication_blocked().await {
crates/scanner/src/scanner/activity.rs
+14 -7
View File
@@ -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<ECStore>,
pub(super) async fn observe_scanner_activity<S>(
storeapi: &Arc<S>,
distributed: bool,
activity_seen: &mut Option<ScannerActivitySnapshot>,
) -> 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<ScannerActivitySnapshot, String> {
let topology_digest = scanner_topology_digest(storeapi);
pub(crate) async fn probe_scanner_activity<S>(storeapi: &S, distributed: bool) -> Result<ScannerActivitySnapshot, String>
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()
crates/scanner/src/scanner/backlog.rs
+31 -15
View File
@@ -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<ScannerPauseBacklogReplic
})
}
async fn load_scanner_pause_backlog(storeapi: Arc<ECStore>) -> Result<LoadedScannerPauseBacklog, String> {
async fn load_scanner_pause_backlog<S>(storeapi: Arc<S>) -> Result<LoadedScannerPauseBacklog, String>
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<ECStore>) -> Result<LoadedScan
select_scanner_pause_backlog_replicas(replicas)
}
async fn write_scanner_pause_backlog_record(
storeapi: Arc<ECStore>,
async fn write_scanner_pause_backlog_record<S>(
storeapi: Arc<S>,
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<ECStore>,
async fn stabilize_scanner_pause_backlog<S>(
storeapi: Arc<S>,
loaded: &LoadedScannerPauseBacklog,
) -> Result<LoadedScannerPauseBacklog, String> {
) -> Result<LoadedScannerPauseBacklog, String>
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<ECStore>,
async fn persist_scanner_pause_backlog<S>(
storeapi: Arc<S>,
loaded: &LoadedScannerPauseBacklog,
ledger: ScannerPauseBacklogLedger,
) -> Result<LoadedScannerPauseBacklog, String> {
) -> Result<LoadedScannerPauseBacklog, String>
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<ECStore>,
pub(super) struct ScannerPauseBacklogController<S: ScannerStorage> {
storeapi: Arc<S>,
loaded: LoadedScannerPauseBacklog,
persistence_disabled: bool,
persistence_retry_at_unix_secs: u64,
}
impl ScannerPauseBacklogController {
pub(super) async fn claim(storeapi: Arc<ECStore>, now: u64) -> Result<Self, String> {
impl<S> ScannerPauseBacklogController<S>
where
S: ScannerStorage,
{
pub(super) async fn claim(storeapi: Arc<S>, now: u64) -> Result<Self, String> {
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<ECStore>, error: String, now: u64) -> Self {
pub(super) fn unavailable(storeapi: Arc<S>, error: String, now: u64) -> Self {
set_runtime_error(Some(error));
let loaded = LoadedScannerPauseBacklog {
ledger: ScannerPauseBacklogLedger::default(),
crates/scanner/src/scanner/heal_info.rs
+11 -6
View File
@@ -62,9 +62,12 @@ pub async fn read_background_heal_info(storeapi: Arc<ECStore>) -> 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<ECStore>) -> BackgroundHealInfoRead {
pub(super) async fn read_background_heal_info_with_epoch<S>(storeapi: Arc<S>) -> 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<ECStore>, info: BackgroundH
save_background_heal_info_for_epoch(storeapi, info, None).await;
}
pub(super) async fn save_background_heal_info_for_epoch(
storeapi: Arc<ECStore>,
pub(super) async fn save_background_heal_info_for_epoch<S>(
storeapi: Arc<S>,
info: BackgroundHealInfo,
expected_epoch: Option<u64>,
) {
) where
S: ScannerStorage,
{
// Skip for ErasureSD setup
if scanner_is_erasure_sd().await {
if storeapi.setup_is_erasure_sd().await {
return;
}
crates/scanner/src/scanner/tests.rs
+2 -1
View File
@@ -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();
crates/scanner/src/scanner_io.rs
+13 -8
View File
@@ -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<S>(
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;
crates/scanner/src/scanner_io/io_cycle.rs
+386 -388
View File
@@ -47,434 +47,432 @@ impl ScannerIOCycle for ECStore {
leader_epoch: u64,
scan_mode: HealScanMode,
) -> Result<ScannerCycleResult> {
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<S>(
store: &S,
ctx: CancellationToken,
budget: Arc<ScannerCycleBudget>,
updates: mpsc::Sender<DataUsageInfo>,
want_cycle: u64,
leader_epoch: u64,
scan_mode: HealScanMode,
) -> Result<ScannerCycleResult>
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::<HashSet<_>>(),
);
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::<HashSet<_>>(),
);
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::<HashSet<_>>() == *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::<HashSet<_>>() == *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<Mutex<Vec<DataUsageCache>>> = Arc::new(Mutex::new(results));
let first_err_mutex: Arc<Mutex<Option<Error>>> = 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<SetDisks> = 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<Mutex<Vec<DataUsageCache>>> = Arc::new(Mutex::new(results));
let first_err_mutex: Arc<Mutex<Option<Error>>> = 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<SetDisks> = 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::<DataUsageCache>(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::<DataUsageCache>(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::<Vec<_>>();
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::<Vec<_>>();
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))
}
crates/scanner/src/storage_api.rs
+282 -14
View File
@@ -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<EcstoreReplicationHealQueueResult> 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<Error = EcstoreErrorType>
+ storage_contracts::NamespaceLocking<Error = EcstoreErrorType, NamespaceLock = NamespaceLockWrapper>
{
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<Notify>;
fn scanner_notification_system(&self) -> Option<Arc<ScannerNotificationSys>>;
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<ScannerBucketListing>;
fn all_set_disks(&self) -> Vec<Arc<EcstoreSetDisks>>;
async fn scanner_pause_backlog_writable_set_disks(&self) -> Vec<Arc<EcstoreSetDisks>>;
#[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<Notify> {
EcstoreStore::scanner_data_movement_changed(self)
}
fn scanner_notification_system(&self) -> Option<Arc<ScannerNotificationSys>> {
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<ScannerBucketListing> {
EcstoreStore::list_bucket_for_scanner(self, opts).await
}
fn all_set_disks(&self) -> Vec<Arc<EcstoreSetDisks>> {
EcstoreStore::all_set_disks(self)
}
async fn scanner_pause_backlog_writable_set_disks(&self) -> Vec<Arc<EcstoreSetDisks>> {
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 = <EcstoreStore as storage_contracts::ObjectIO>::ObjectOptions;
type ObjectInfo = <EcstoreStore as storage_contracts::ObjectIO>::ObjectInfo;
type GetObjectReader = <EcstoreStore as storage_contracts::ObjectIO>::GetObjectReader;
type PutObjectReader = <EcstoreStore as storage_contracts::ObjectIO>::PutObjectReader;
async fn get_object_reader(
&self,
_bucket: &str,
_object: &str,
_range: Option<Self::RangeSpec>,
_h: Self::HeaderMap,
_opts: &Self::ObjectOptions,
) -> Result<Self::GetObjectReader, Self::Error> {
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<Self::ObjectInfo, Self::Error> {
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<storage_contracts::BucketInfo, Self::Error> {
Err(EcstoreErrorType::other("fake scanner storage cannot read buckets"))
}
async fn list_bucket(
&self,
_opts: &storage_contracts::BucketOptions,
) -> Result<Vec<storage_contracts::BucketInfo>, 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<Self::NamespaceLock, Self::Error> {
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: <EcstoreStore as storage_contracts::ObjectIO>::ObjectOptions,
) -> EcstoreResultType<<EcstoreStore as storage_contracts::ObjectIO>::ObjectInfo> {
Err(EcstoreErrorType::other("fake scanner storage cannot delete objects"))
}
async fn scanner_data_usage_publication_admission(&self) -> Option<crate::ScannerDataUsagePublicationAdmission> {
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<Notify> {
Arc::new(Notify::new())
}
fn scanner_notification_system(&self) -> Option<Arc<ScannerNotificationSys>> {
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<ScannerBucketListing> {
Err(EcstoreErrorType::other("fake scanner storage cannot list scanner buckets"))
}
fn all_set_disks(&self) -> Vec<Arc<EcstoreSetDisks>> {
Vec::new()
}
async fn scanner_pause_backlog_writable_set_disks(&self) -> Vec<Arc<EcstoreSetDisks>> {
Vec::new()
}
fn scanner_observed_probe_store_key(&self) -> usize {
0
}
}
fn assert_scanner_storage<S: ScannerStorage>() {}
#[test]
fn scanner_storage_contract_accepts_fake_without_ecstore() {
assert_scanner_storage::<FakeScannerStorage>();
}
}
docs/architecture/crate-boundaries.md
+2 -2
View File
@@ -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.