From 0e6ee3bf62ac772fe554d62302f91371a4b6509d Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 3 Sep 2026 17:02:43 +0800 Subject: [PATCH] feat(scanner): coordinate usage and workload boundaries (#7093) * test(scanner): wire usage and heal rebuild gates * docs(scanner): define usage authority protocol * docs(heal): clarify scanner and ecstore boundaries * refactor(scanner): split metrics from contracts * feat(scanner): use shared workload snapshots * fix(ecstore): recheck capacity before decommission drain --- .config/e2e-full-selection.txt | 4 +- .config/e2e-smoke-selection.txt | 2 +- .config/nextest.toml | 25 ++- ARCHITECTURE.md | 4 +- Cargo.lock | 15 +- Cargo.toml | 4 +- crates/ecstore/Cargo.toml | 2 +- .../lifecycle/bucket_lifecycle_audit.rs | 2 +- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 4 +- .../src/bucket/lifecycle/replication_sink.rs | 4 +- crates/ecstore/src/bucket/quota/checker.rs | 10 +- crates/ecstore/src/core/pools.rs | 44 ++++- crates/ecstore/src/object_api/types.rs | 2 +- .../ecstore/src/services/metrics_realtime.rs | 52 +++--- crates/ecstore/src/set_disk/ops/object.rs | 6 +- crates/ecstore/src/store/init.rs | 4 +- crates/ecstore/src/store/object.rs | 2 +- crates/ecstore/src/store/rebalance.rs | 4 +- crates/lifecycle/Cargo.toml | 2 +- crates/lifecycle/src/core.rs | 2 +- crates/lifecycle/src/evaluator.rs | 4 +- crates/lifecycle/src/lib.rs | 2 +- crates/obs/Cargo.toml | 2 +- crates/obs/src/metrics/collectors/scanner.rs | 2 +- crates/obs/src/metrics/stats_collector.rs | 8 +- crates/scanner-contracts/Cargo.toml | 14 +- crates/scanner-contracts/src/lib.rs | 6 - crates/scanner-metrics/Cargo.toml | 45 +++++ .../src/init_time.rs | 0 .../src/last_minute.rs | 0 crates/scanner-metrics/src/lib.rs | 19 ++ .../src/metrics.rs | 0 crates/scanner/Cargo.toml | 3 +- crates/scanner/src/lib.rs | 4 +- crates/scanner/src/remote_scanner/stream.rs | 2 +- crates/scanner/src/scanner.rs | 2 +- crates/scanner/src/scanner_folder.rs | 2 +- crates/scanner/src/scanner_io.rs | 2 +- crates/scanner/src/scanner_io/guards.rs | 8 +- crates/scanner/src/scanner_io/io_disk.rs | 4 +- crates/scanner/src/scanner_io/tests.rs | 41 +++++ crates/scanner/src/sleeper.rs | 34 ++-- crates/scanner/src/workload_admission.rs | 165 ++++++++++++++++++ docs/architecture/README.md | 2 + docs/architecture/crate-boundaries.md | 14 ++ docs/architecture/heal-concurrency-model.md | 2 + .../scanner-usage-authority-decision.md | 24 +++ .../architecture/scanner-usage-publication.md | 129 ++++++++++++++ .../workload-admission-contracts.md | 5 +- docs/testing/ci-gates.md | 8 + rustfs/Cargo.toml | 2 +- rustfs/src/admin/handlers/scanner.rs | 6 +- rustfs/src/admin/storage_api.rs | 2 +- rustfs/src/app/storage_api.rs | 4 +- rustfs/src/cluster_snapshot.rs | 2 +- rustfs/src/startup_background.rs | 1 + rustfs/src/startup_runtime_sources.rs | 2 +- 57 files changed, 632 insertions(+), 134 deletions(-) create mode 100644 crates/scanner-metrics/Cargo.toml rename crates/{scanner-contracts => scanner-metrics}/src/init_time.rs (100%) rename crates/{scanner-contracts => scanner-metrics}/src/last_minute.rs (100%) create mode 100644 crates/scanner-metrics/src/lib.rs rename crates/{scanner-contracts => scanner-metrics}/src/metrics.rs (100%) create mode 100644 crates/scanner/src/workload_admission.rs create mode 100644 docs/architecture/scanner-usage-authority-decision.md create mode 100644 docs/architecture/scanner-usage-publication.md diff --git a/.config/e2e-full-selection.txt b/.config/e2e-full-selection.txt index 204565e83..dc4d9355b 100644 --- a/.config/e2e-full-selection.txt +++ b/.config/e2e-full-selection.txt @@ -1,2 +1,2 @@ -sha256-darwin=21f2a5484bb10bed1f11bd6397fe80f20d60a870e4ed5016a1f3acbdb4e6f735 -sha256-linux=29fd001e70d0114b59d65dd2a516c1f96a5fea1d6bb8fb315550a27f94d7b39b +sha256-darwin=bb8c16cd63a94ff5e5e400d891ccc980710529dd5cb13a5a0c955c6c313bccd9 +sha256-linux=e9a8d64e73f627c4d26c236dbbba690c9ee03a9e26d42a4244515b4439365535 diff --git a/.config/e2e-smoke-selection.txt b/.config/e2e-smoke-selection.txt index cda5ece21..0a73fc814 100644 --- a/.config/e2e-smoke-selection.txt +++ b/.config/e2e-smoke-selection.txt @@ -1 +1 @@ -sha256=d90fa941cb737fe6fcf39ca13d45440183e77a7e047880c025af7e6bfe1d0517 +sha256=a2542dc86bbff56b2177efc621785c56fa7e8d813b209b7d935e1e41a9f0ad15 diff --git a/.config/nextest.toml b/.config/nextest.toml index e85a10ebc..cc40c78f1 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -407,10 +407,15 @@ test-group = 'ecstore-serial-flaky' # harness_self}_test` stays in e2e-full and the fault / concurrency / # real-source modules stay in e2e-nightly; this is an allowlist, not a module # clause, so a new ODM test never lands here silently. +# +# Scanner authoritative usage publication (backlog#2213): data_usage_test is +# the PR-lane e2e coverage for scanner usage snapshots consumed by quota and +# admin surfaces. It uses the same single-node, random-port, isolated-temp-dir +# fixture as the existing smoke modules. [profile.e2e-smoke] default-filter = """ package(e2e_test) & ( - test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|list_buckets_auth|list_buckets_iam_filter|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_object_tagging|copy_source_invalid_date|content_encoding|compression|multipart_storage_class|storage_class_capability|ssec_copy|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth|notification_webhook|tls_hot_reload|console_smoke|admin_iam_crud|admin_pools|sts_query_compat)_test::|^fake_s3_target::/) + test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|list_buckets_auth|list_buckets_iam_filter|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_object_tagging|copy_source_invalid_date|content_encoding|compression|multipart_storage_class|storage_class_capability|ssec_copy|anonymous_access|bucket_policy_check|presigned_negative|negative_sigv4|admin_auth|notification_webhook|tls_hot_reload|console_smoke|admin_iam_crud|admin_pools|sts_query_compat|data_usage)_test::|^fake_s3_target::/) | test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/) | test(/^reliant::lifecycle::/) | test(/^reliant::tiering::/) @@ -488,12 +493,16 @@ path = "junit.xml" # --------------------------------------------------------------------------- # e2e-nightly profile — destructive multi-process cluster fault domains # --------------------------------------------------------------------------- -# These seven modules are deliberately outside e2e-full's merge budget. Each +# These eight modules are deliberately outside e2e-full's merge budget. Each # starts a real multi-process or multi-disk topology and exercises node/disk # loss, quorum, cleanup, notification fan-in, or admin-timeout behavior. The # consolidated nightly workflow runs them serially to avoid resource # starvation; failures are never retried. # +# heal_erasure_disk_rebuild_test also runs in e2e-full so core heal rebuild +# regressions are caught no later than the merge/main lane. It remains here for +# nightly serial coverage with the other cluster fault domains. +# # On-demand migration (backlog#2158 ODM-11) joins by the second clause: the # fault matrix waits out the 30 s circuit-breaker window, the concurrency # matrix drives 100-deep bursts, and the real-source cases start a second @@ -542,11 +551,13 @@ path = "junit.xml" # The filter is "the whole e2e_test crate MINUS the sets owned by other lanes": # * protocols:: — FTPS/SFTP/WebDAV, run from the dedicated protocol profile # with one worker because the suite owns fixed ports. -# * the 7 cluster suites that spin up a RustFSTestClusterEnvironment +# * cluster suites that spin up a RustFSTestClusterEnvironment # (cluster_concurrency, cluster_multidrive_pool, stale_multipart_cleanup_cluster, -# namespace_lock_quorum, heal_erasure_disk_rebuild, admin_timeout_regression, -# object_lambda) — too heavy for the merge budget; they run in the -# e2e-nightly serial cluster-fault lane. +# namespace_lock_quorum, admin_timeout_regression, object_lambda) — too +# heavy for the merge budget; they run in the e2e-nightly serial +# cluster-fault lane. heal_erasure_disk_rebuild is intentionally not +# excluded here because backlog#2213 promotes core heal rebuild coverage to +# this merge/main lane while retaining nightly coverage. # * replication_extension_test — repl-1 already splits it into the PR # `e2e-smoke` (20 fast) and `e2e-repl-nightly` (56 slow) lanes and reserves # it for those, so e2e-full does not double-run it. @@ -562,7 +573,7 @@ path = "junit.xml" default-filter = """ package(e2e_test) & !test(/^protocols::/) - & !test(/^(admin_timeout_regression_test|cluster_concurrency_test|cluster_multidrive_pool_test|degraded_listing_availability_test|heal_erasure_disk_rebuild_test|namespace_lock_quorum_test|object_lambda_test|stale_multipart_cleanup_cluster_test)::/) + & !test(/^(admin_timeout_regression_test|cluster_concurrency_test|cluster_multidrive_pool_test|degraded_listing_availability_test|namespace_lock_quorum_test|object_lambda_test|stale_multipart_cleanup_cluster_test)::/) & !test(/^replication_extension_test::/) & !test(/^replication_target_matrix_test::/) & !test(/^on_demand_migration::(concurrency_test|fault_test|real_source_test)::/) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 8e4f7c043..da8f386a0 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`, `utils` | Shared configuration, data-usage models, heal/scanner domain contracts, utilities, and checksums. | +| 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. | | 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. | @@ -135,7 +135,7 @@ default build (lifecycle: `crates/ecstore/src/bucket/replication/replication_state.rs`) — a naming collision, not copies; renaming is tracked in rustfs/backlog#1847. - `LastMinuteLatency` has two deliberately different implementations: the - per-second bucketed accumulator in `crates/scanner-contracts/src/last_minute.rs` and + per-second bucketed accumulator in `crates/scanner-metrics/src/last_minute.rs` and the in-memory endpoint-health sample tracker in `crates/ecstore/src/bucket/bucket_target_sys.rs` (its doc comment explains why it stays local). diff --git a/Cargo.lock b/Cargo.lock index d227106ac..e9195be56 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9517,7 +9517,7 @@ dependencies = [ "rustfs-s3select-api", "rustfs-s3select-query", "rustfs-scanner", - "rustfs-scanner-contracts", + "rustfs-scanner-metrics", "rustfs-security-governance", "rustfs-signer", "rustfs-storage-api", @@ -9763,7 +9763,7 @@ dependencies = [ "rustfs-rio-v2", "rustfs-s3-client", "rustfs-s3-types", - "rustfs-scanner-contracts", + "rustfs-scanner-metrics", "rustfs-storage-api", "rustfs-uring", "rustfs-utils", @@ -10119,7 +10119,7 @@ dependencies = [ "proptest", "rustfs-config", "rustfs-replication", - "rustfs-scanner-contracts", + "rustfs-scanner-metrics", "rustfs-storage-api", "s3s", "serial_test", @@ -10317,7 +10317,7 @@ dependencies = [ "rustfs-iam", "rustfs-io-metrics", "rustfs-notify", - "rustfs-scanner-contracts", + "rustfs-scanner-metrics", "rustfs-security-governance", "rustfs-storage-api", "rustfs-utils", @@ -10674,6 +10674,7 @@ dependencies = [ "rand 0.10.2", "rmp-serde", "rustfs-common", + "rustfs-concurrency", "rustfs-config", "rustfs-credentials", "rustfs-data-usage", @@ -10682,7 +10683,7 @@ dependencies = [ "rustfs-heal-contracts", "rustfs-lock", "rustfs-s3-types", - "rustfs-scanner-contracts", + "rustfs-scanner-metrics", "rustfs-storage-api", "rustfs-utils", "s3s", @@ -10704,6 +10705,10 @@ dependencies = [ [[package]] name = "rustfs-scanner-contracts" version = "1.0.0-rc.5" + +[[package]] +name = "rustfs-scanner-metrics" +version = "1.0.0-rc.5" dependencies = [ "chrono", "jiff", diff --git a/Cargo.toml b/Cargo.toml index 20cc7048e..fc7b44766 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -52,7 +52,8 @@ 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 metrics and cycle contracts + "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 "crates/signer", # client signer @@ -94,6 +95,7 @@ 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" } rustfs-common = { path = "crates/common", version = "1.0.0-rc.5" } diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 9358e1a54..50aa7c1d3 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -130,7 +130,7 @@ rustfs-concurrency.workspace = true rustfs-credentials = { workspace = true } rustfs-common.workspace = true rustfs-heal-contracts.workspace = true -rustfs-scanner-contracts.workspace = true +rustfs-scanner-metrics.workspace = true rustfs-policy.workspace = true rustfs-protos.workspace = true rustfs-replication.workspace = true diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_audit.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_audit.rs index 06ea063a7..e7dde4652 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_audit.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_audit.rs @@ -17,7 +17,7 @@ use crate::bucket::lifecycle::lifecycle; use crate::object_api::ObjectInfo; use crate::services::event_notification::{EventArgs, send_event}; use rustfs_s3_types::EventName; -use rustfs_scanner_contracts::metrics::IlmAction; +use rustfs_scanner_metrics::metrics::IlmAction; const LIFECYCLE_EXPIRY_USER_AGENT: &str = "Internal: [ILM-Expiry]"; const LIFECYCLE_TRANSITION_USER_AGENT: &str = "Internal: [ILM-Transition]"; diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 6b637cfac..0ff7edc8b 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -84,7 +84,7 @@ use rustfs_data_usage::TierStats; use rustfs_filemeta::{ FileInfo, FileInfoOpts, NULL_VERSION_ID, RestoreStatusOps, TRANSITION_COMPLETE, get_file_info, is_restored_object_on_disk, }; -use rustfs_scanner_contracts::metrics::{ +use rustfs_scanner_metrics::metrics::{ IlmAction, Metrics, ScannerLifecycleExpiryStateUpdate, ScannerLifecycleTransitionStateUpdate, global_metrics, }; use rustfs_utils::{ @@ -5596,7 +5596,7 @@ mod tests { use rustfs_filemeta::{FileInfo, FileMeta}; #[cfg(feature = "test-util")] use rustfs_s3_client::transition_api::ReaderImpl; - use rustfs_scanner_contracts::metrics::{IlmAction, global_metrics}; + use rustfs_scanner_metrics::metrics::{IlmAction, global_metrics}; use s3s::dto::{ BucketLifecycleConfiguration, DefaultRetention, ExpirationStatus, LifecycleExpiration, LifecycleRule, MetadataEntry, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRetentionMode, ObjectLockRule, OutputLocation, RestoreRequest, diff --git a/crates/ecstore/src/bucket/lifecycle/replication_sink.rs b/crates/ecstore/src/bucket/lifecycle/replication_sink.rs index 70b700883..042759176 100644 --- a/crates/ecstore/src/bucket/lifecycle/replication_sink.rs +++ b/crates/ecstore/src/bucket/lifecycle/replication_sink.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use rustfs_scanner_contracts::metrics::IlmAction; +use rustfs_scanner_metrics::metrics::IlmAction; use crate::bucket::lifecycle::lifecycle::ObjectOpts; use crate::bucket::replication::ReplicationLifecycleBridge; @@ -77,7 +77,7 @@ mod tests { use crate::bucket::replication::{DeleteReplicationConfigSnapshot, ReplicationObjectBridge}; use crate::object_api::{ObjectInfo, ObjectOptions}; use crate::storage_api_contracts::object::ObjectToDelete; - use rustfs_scanner_contracts::metrics::IlmAction; + use rustfs_scanner_metrics::metrics::IlmAction; use s3s::dto::{ BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, diff --git a/crates/ecstore/src/bucket/quota/checker.rs b/crates/ecstore/src/bucket/quota/checker.rs index 44a113652..8e313c09a 100644 --- a/crates/ecstore/src/bucket/quota/checker.rs +++ b/crates/ecstore/src/bucket/quota/checker.rs @@ -16,7 +16,7 @@ use super::{BucketQuota, QuotaCheckResult, QuotaError, QuotaOperation}; use crate::bucket::metadata_sys::{BucketMetadataSys, update, update_if_incarnation}; use crate::data_usage::get_bucket_usage_memory; use rustfs_config::QUOTA_CONFIG_FILE; -use rustfs_scanner_contracts::metrics::Metric; +use rustfs_scanner_metrics::metrics::Metric; use std::sync::Arc; use std::time::Instant; use time::OffsetDateTime; @@ -120,9 +120,9 @@ impl QuotaChecker { let duration = start_time.elapsed(); // inc_time is now a plain fn (not async) — no .await needed. - rustfs_scanner_contracts::metrics::Metrics::inc_time(Metric::QuotaCheck, duration); + rustfs_scanner_metrics::metrics::Metrics::inc_time(Metric::QuotaCheck, duration); if !allowed { - rustfs_scanner_contracts::metrics::Metrics::inc_time(Metric::QuotaViolation, duration); + rustfs_scanner_metrics::metrics::Metrics::inc_time(Metric::QuotaViolation, duration); } Ok(result) @@ -185,7 +185,7 @@ impl QuotaChecker { .await .map_err(QuotaError::StorageError)?; - rustfs_scanner_contracts::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed()); + rustfs_scanner_metrics::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed()); Ok(updated_at) } @@ -206,7 +206,7 @@ impl QuotaChecker { } .map_err(QuotaError::StorageError)?; - rustfs_scanner_contracts::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed()); + rustfs_scanner_metrics::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed()); Ok(updated_at) } diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 956b3c3d1..4973f6831 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -13504,7 +13504,9 @@ impl ECStore { } return Ok(()); } - let result = self.decommission_in_background(rx.clone(), idx, entry_budget).await; + let result = self + .decommission_in_background(rx.clone(), idx, generation, entry_budget) + .await; if let Err(err) = &result && (is_decommission_capacity_blocked_error(err) || is_decommission_target_capacity_error(err)) @@ -13989,8 +13991,10 @@ impl ECStore { self: &Arc, rx: CancellationToken, idx: usize, + generation: OffsetDateTime, entry_budget: Arc, ) -> Result<()> { + self.ensure_decommission_runtime_capacity_available(idx, generation).await?; let pool = get_by_index(self.pools.as_slice(), idx, "load decommission background pool")?.clone(); let pending = { @@ -17253,6 +17257,44 @@ mod tests { ); } + #[tokio::test] + #[serial_test::serial] + async fn decommission_worker_rechecks_runtime_capacity_before_empty_background_completion() { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + let enough = vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 30, 30), + DecommissionPoolCapacityInfo::for_test(1, layout, 60, 60, 0), + ]; + set_decommission_capacity_info_overrides_for_test(store.id, vec![enough.clone()]); + store + .save_current_pool_meta_for_decommission_start(&[0], Vec::new()) + .await + .expect("the initial reservation should be activated"); + + let shortage = vec![enough[0], DecommissionPoolCapacityInfo::for_test(1, layout, 59, 60, 1)]; + set_decommission_capacity_info_overrides_for_test(store.id, vec![shortage]); + let canceler = DecommissionCanceler::new(CancellationToken::new()); + store.decommission_cancelers.write().await[0] = Some(canceler.clone()); + store + .do_decommission_in_routine(canceler, 0, Arc::new(Semaphore::new(1))) + .await + .expect("runtime capacity shortage should pause the worker before background completion"); + + let local = store.pool_meta.read().await; + let info = local.pools[0] + .decommission + .as_ref() + .expect("the blocked decommission state should remain present"); + assert!(!info.complete && !info.failed && !info.canceled); + assert!(info.capacity_blocked_reason.is_some()); + assert!( + info.capacity_reservation + .as_ref() + .is_some_and(DecommissionCapacityReservation::active) + ); + } + fn pool_meta_replica_test_meta(cmd_line: &str) -> PoolMeta { PoolMeta { version: POOL_META_VERSION, diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 62a45e075..85e270679 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -286,7 +286,7 @@ pub struct QuotaAdmission { pub struct LifecycleDeleteAllRequest { pub(crate) version_id: Option, pub(crate) delete_marker: bool, - pub(crate) action: rustfs_scanner_contracts::metrics::IlmAction, + pub(crate) action: rustfs_scanner_metrics::metrics::IlmAction, pub(crate) rule_id: String, pub(crate) phase: LifecycleDeleteAllPhase, } diff --git a/crates/ecstore/src/services/metrics_realtime.rs b/crates/ecstore/src/services/metrics_realtime.rs index c6056414d..a22400fd9 100644 --- a/crates/ecstore/src/services/metrics_realtime.rs +++ b/crates/ecstore/src/services/metrics_realtime.rs @@ -32,7 +32,7 @@ use rustfs_madmin::metrics::{ ScannerSourceCycleSnapshot as MadminScannerSourceCycleSnapshot, ScannerSourceWorkSnapshot as MadminScannerSourceWorkSnapshot, ScannerUsageFreshnessSnapshot as MadminScannerUsageFreshnessSnapshot, TimedAction as MadminTimedAction, }; -use rustfs_scanner_contracts::metrics::global_metrics; +use rustfs_scanner_metrics::metrics::global_metrics; use rustfs_utils::os::get_drive_stats; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; @@ -82,7 +82,7 @@ fn unix_millis_to_jiff_timestamp(millis: u64, fallback: Timestamp) -> Timestamp } } -fn to_madmin_scanner_metrics(metrics: rustfs_scanner_contracts::metrics::ScannerMetricsReport) -> MadminScannerMetrics { +fn to_madmin_scanner_metrics(metrics: rustfs_scanner_metrics::metrics::ScannerMetricsReport) -> MadminScannerMetrics { MadminScannerMetrics { collected_at: metrics.collected_at, current_cycle: metrics.current_cycle, @@ -565,7 +565,7 @@ async fn collect_local_disks_metrics(disks: &HashSet) -> HashMap String { if name.ends_with("_total") { diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index b3ed51655..36b4641d7 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -45,7 +45,7 @@ use rustfs_io_metrics::{ ProcessResourceSnapshot, ProcessSampler, ProcessStatusSnapshot, ProcessSystemSnapshot, s3_op_metrics_snapshot, snapshot_process_resource_and_system, snapshot_process_resource_and_system_with, }; -use rustfs_scanner_contracts::metrics::{ +use rustfs_scanner_metrics::metrics::{ ScannerActiveBucketDriveSnapshot, ScannerBucketDriveResultSnapshot, ScannerMetricsReport, ScannerSourceWorkSnapshot, global_metrics, }; @@ -1679,7 +1679,7 @@ pub async fn collect_compression_cluster_stats() -> Option Self { + pub(super) fn new(source: rustfs_scanner_metrics::metrics::ScannerWorkSource, bucket: &str, drive: &str) -> Self { Self { failed: true, source, @@ -245,7 +245,7 @@ pub(super) fn scanner_concurrency_limit(configured: usize, available: usize) -> return 0; } - if crate::current_foreground_read_activity() > 0 { + if crate::workload_admission::foreground_workload_activity() > 0 { return 1; } @@ -285,7 +285,7 @@ pub(super) fn scanner_task_join_error(stage: &str, err: tokio::task::JoinError) #[cfg(test)] mod tests { use super::*; - use rustfs_scanner_contracts::metrics::{ScannerWorkSource, global_metrics}; + use rustfs_scanner_metrics::metrics::{ScannerWorkSource, global_metrics}; use tokio::sync::oneshot; fn active_bucket_drive_count(source: ScannerWorkSource, bucket: &str, drive: &str) -> u64 { diff --git a/crates/scanner/src/scanner_io/io_disk.rs b/crates/scanner/src/scanner_io/io_disk.rs index 1034d929e..2bb55b55c 100644 --- a/crates/scanner/src/scanner_io/io_disk.rs +++ b/crates/scanner/src/scanner_io/io_disk.rs @@ -161,8 +161,8 @@ impl ScannerIODisk for Disk { let bucket = cache.info.name.clone(); let disk_path = self.path().to_string_lossy().to_string(); let source = match scan_mode { - HealScanMode::Deep => rustfs_scanner_contracts::metrics::ScannerWorkSource::Bitrot, - HealScanMode::Normal | HealScanMode::Unknown => rustfs_scanner_contracts::metrics::ScannerWorkSource::Usage, + HealScanMode::Deep => rustfs_scanner_metrics::metrics::ScannerWorkSource::Bitrot, + HealScanMode::Normal | HealScanMode::Unknown => rustfs_scanner_metrics::metrics::ScannerWorkSource::Usage, }; global_metrics().record_scan_bucket_drive_start(source, &bucket, &disk_path); let mut failure_guard = BucketDriveFailureGuard::new(source, &bucket, &disk_path); diff --git a/crates/scanner/src/scanner_io/tests.rs b/crates/scanner/src/scanner_io/tests.rs index df86fdc1d..2b086aa5e 100644 --- a/crates/scanner/src/scanner_io/tests.rs +++ b/crates/scanner/src/scanner_io/tests.rs @@ -26,12 +26,32 @@ use crate::{ ScannerPutObjReader, UNKNOWN_TIER, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx, new_disk, path2_bucket_object_with_base_path, }; +use rustfs_concurrency::{ + AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadAdmissionSnapshotProvider, + WorkloadClass, +}; use rustfs_filemeta::FileInfo; use serial_test::serial; +use std::sync::Arc; use temp_env::with_var; use time::OffsetDateTime; use uuid::Uuid; +#[derive(Clone)] +struct FixedWorkloadProvider { + snapshot: WorkloadAdmissionRegistrySnapshot, +} + +impl WorkloadAdmissionSnapshotProvider for FixedWorkloadProvider { + fn workload_admission_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot { + self.snapshot.clone() + } +} + +fn install_scanner_workload_provider(snapshot: WorkloadAdmissionRegistrySnapshot) { + crate::set_scanner_workload_admission_snapshot_provider(Arc::new(FixedWorkloadProvider { snapshot })); +} + fn bucket_info(name: &str) -> BucketInfo { BucketInfo { name: name.to_string(), @@ -1079,6 +1099,7 @@ async fn bucket_cache_pending_heal_reaches_cycle_maintenance_state() { #[serial] fn scanner_concurrency_limit_preserves_available_when_unconfigured() { crate::reset_foreground_read_activity_for_test(); + crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); assert_eq!(scanner_concurrency_limit(0, 4), 4); } @@ -1086,6 +1107,7 @@ fn scanner_concurrency_limit_preserves_available_when_unconfigured() { #[serial] fn scanner_concurrency_limit_caps_to_configured_value() { crate::reset_foreground_read_activity_for_test(); + crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); assert_eq!(scanner_concurrency_limit(2, 4), 2); } @@ -1093,6 +1115,7 @@ fn scanner_concurrency_limit_caps_to_configured_value() { #[serial] fn scanner_concurrency_limit_never_exceeds_available_work() { crate::reset_foreground_read_activity_for_test(); + crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); assert_eq!(scanner_concurrency_limit(8, 4), 4); } @@ -1100,6 +1123,7 @@ fn scanner_concurrency_limit_never_exceeds_available_work() { #[serial] fn scanner_concurrency_limit_handles_no_available_work() { crate::reset_foreground_read_activity_for_test(); + crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); assert_eq!(scanner_concurrency_limit(2, 0), 0); } @@ -1107,16 +1131,33 @@ fn scanner_concurrency_limit_handles_no_available_work() { #[serial] fn scanner_concurrency_limit_yields_to_foreground_reads() { crate::reset_foreground_read_activity_for_test(); + crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); crate::set_foreground_read_activity(8); assert_eq!(scanner_concurrency_limit(0, 4), 1); assert_eq!(scanner_concurrency_limit(3, 4), 1); crate::reset_foreground_read_activity_for_test(); } +#[test] +#[serial] +fn scanner_concurrency_limit_yields_to_shared_foreground_pressure() { + crate::reset_foreground_read_activity_for_test(); + crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); + install_scanner_workload_provider(WorkloadAdmissionRegistrySnapshot::new(vec![ + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, AdmissionState::Open).with_counts(Some(2), None, Some(16)), + ])); + + assert_eq!(scanner_concurrency_limit(0, 4), 1); + assert_eq!(scanner_concurrency_limit(3, 4), 1); + + crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); +} + #[test] #[serial] fn scanner_concurrency_limit_yields_to_streaming_reads() { crate::reset_foreground_read_activity_for_test(); + crate::workload_admission::clear_scanner_workload_admission_snapshot_provider_for_test(); let _guard = crate::ForegroundReadGuard::new(); assert_eq!(scanner_concurrency_limit(0, 4), 1); diff --git a/crates/scanner/src/sleeper.rs b/crates/scanner/src/sleeper.rs index 8546cc99a..2af363f7f 100644 --- a/crates/scanner/src/sleeper.rs +++ b/crates/scanner/src/sleeper.rs @@ -20,7 +20,7 @@ use rustfs_config::{ DEFAULT_SCANNER_IDLE_MODE, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, ENV_SCANNER_IDLE_MODE, ENV_SCANNER_SPEED, ENV_SCANNER_YIELD_EVERY_N_OBJECTS, ScannerSpeed, }; -use rustfs_scanner_contracts::metrics::global_metrics; +use rustfs_scanner_metrics::metrics::global_metrics; use tokio::time::Duration; const MIN_SLEEP: Duration = Duration::from_millis(1); @@ -29,8 +29,8 @@ const SCANNER_SPEED_FAST: u8 = 1; const SCANNER_SPEED_DEFAULT: u8 = 2; const SCANNER_SPEED_SLOW: u8 = 3; const SCANNER_SPEED_SLOWEST: u8 = 4; -const FOREGROUND_READ_BACKOFF_PER_REQUEST_MS: u64 = 10; -const FOREGROUND_READ_BACKOFF_MAX_MS: u64 = 250; +const FOREGROUND_WORKLOAD_BACKOFF_PER_REQUEST_MS: u64 = 10; +const FOREGROUND_WORKLOAD_BACKOFF_MAX_MS: u64 = 250; static SCANNER_DEFAULT_SPEED_PRESET: AtomicU8 = AtomicU8::new(SCANNER_SPEED_DEFAULT); @@ -78,15 +78,15 @@ pub(crate) fn scanner_yield_every_n_objects() -> u64 { rustfs_utils::get_env_u64(ENV_SCANNER_YIELD_EVERY_N_OBJECTS, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS) } -fn foreground_read_backoff_duration(active_reads: u64) -> Duration { - if active_reads == 0 { +fn foreground_workload_backoff_duration(active_foreground_workloads: u64) -> Duration { + if active_foreground_workloads == 0 { return Duration::ZERO; } Duration::from_millis( - active_reads - .saturating_mul(FOREGROUND_READ_BACKOFF_PER_REQUEST_MS) - .min(FOREGROUND_READ_BACKOFF_MAX_MS), + active_foreground_workloads + .saturating_mul(FOREGROUND_WORKLOAD_BACKOFF_PER_REQUEST_MS) + .min(FOREGROUND_WORKLOAD_BACKOFF_MAX_MS), ) } @@ -147,14 +147,15 @@ impl DynamicSleeper { } let (factor, max_sleep) = self.read_params(); if factor == 0.0 || max_sleep.is_zero() { - let foreground_sleep = foreground_read_backoff_duration(crate::current_foreground_read_activity()); + let foreground_sleep = + foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity()); if !foreground_sleep.is_zero() { tokio::time::sleep(foreground_sleep).await; global_metrics().record_scanner_throttle_sleep(foreground_sleep); } return; } - let foreground_sleep = foreground_read_backoff_duration(crate::current_foreground_read_activity()); + let foreground_sleep = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity()); let sleep_dur = Duration::from_secs_f64(MIN_SLEEP.as_secs_f64() * factor) .min(max_sleep) .max(foreground_sleep); @@ -235,7 +236,8 @@ impl SleepTimer { } let (factor, max_sleep) = self.sleeper.read_params(); if factor == 0.0 || max_sleep.is_zero() { - let foreground_sleep = foreground_read_backoff_duration(crate::current_foreground_read_activity()); + let foreground_sleep = + foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity()); if !foreground_sleep.is_zero() { tokio::time::sleep(foreground_sleep).await; global_metrics().record_scanner_throttle_sleep(foreground_sleep); @@ -243,7 +245,7 @@ impl SleepTimer { return; } let elapsed = self.start.elapsed(); - let foreground_sleep = foreground_read_backoff_duration(crate::current_foreground_read_activity()); + let foreground_sleep = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity()); let sleep_dur = Duration::from_secs_f64(elapsed.as_secs_f64() * factor) .max(MIN_SLEEP) .min(max_sleep) @@ -304,10 +306,10 @@ mod tests { } #[test] - fn foreground_read_backoff_is_capped() { - assert_eq!(foreground_read_backoff_duration(0), Duration::ZERO); - assert_eq!(foreground_read_backoff_duration(1), Duration::from_millis(10)); - assert_eq!(foreground_read_backoff_duration(80), Duration::from_millis(250)); + fn foreground_workload_backoff_is_capped() { + assert_eq!(foreground_workload_backoff_duration(0), Duration::ZERO); + assert_eq!(foreground_workload_backoff_duration(1), Duration::from_millis(10)); + assert_eq!(foreground_workload_backoff_duration(80), Duration::from_millis(250)); } #[test] diff --git a/crates/scanner/src/workload_admission.rs b/crates/scanner/src/workload_admission.rs new file mode 100644 index 000000000..06b008b93 --- /dev/null +++ b/crates/scanner/src/workload_admission.rs @@ -0,0 +1,165 @@ +// 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. + +use std::sync::{Arc, LazyLock, RwLock}; + +use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshotProvider, WorkloadClass}; + +type WorkloadSnapshotProviderRef = Arc; + +static SCANNER_WORKLOAD_ADMISSION_PROVIDER: LazyLock>> = + LazyLock::new(|| RwLock::new(None)); + +pub fn set_scanner_workload_admission_snapshot_provider(provider: WorkloadSnapshotProviderRef) { + *SCANNER_WORKLOAD_ADMISSION_PROVIDER + .write() + .unwrap_or_else(|err| err.into_inner()) = Some(provider); +} + +fn scanner_workload_admission_snapshot_provider() -> Option { + SCANNER_WORKLOAD_ADMISSION_PROVIDER + .read() + .unwrap_or_else(|err| err.into_inner()) + .clone() +} + +#[cfg(test)] +pub(crate) fn clear_scanner_workload_admission_snapshot_provider_for_test() { + *SCANNER_WORKLOAD_ADMISSION_PROVIDER + .write() + .unwrap_or_else(|err| err.into_inner()) = None; +} + +pub(crate) fn foreground_workload_activity() -> u64 { + let local_activity = crate::current_foreground_read_activity(); + let Some(provider) = scanner_workload_admission_snapshot_provider() else { + return local_activity; + }; + + local_activity.max(foreground_activity_from_snapshot(&provider.workload_admission_snapshot())) +} + +fn foreground_activity_from_snapshot(snapshot: &WorkloadAdmissionRegistrySnapshot) -> u64 { + [WorkloadClass::ForegroundRead, WorkloadClass::ForegroundWrite] + .into_iter() + .filter_map(|class| snapshot.get(class)) + .map(|entry| { + entry + .active + .or_else(|| { + matches!(entry.state, AdmissionState::Saturated).then(|| entry.limit.filter(|limit| *limit > 0).unwrap_or(1)) + }) + .map(usize_to_u64_saturated) + .unwrap_or(0) + }) + .max() + .unwrap_or(0) +} + +fn usize_to_u64_saturated(value: usize) -> u64 { + u64::try_from(value).unwrap_or(u64::MAX) +} + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshot}; + use serial_test::serial; + + #[derive(Clone)] + struct FixedWorkloadProvider { + snapshot: WorkloadAdmissionRegistrySnapshot, + } + + impl WorkloadAdmissionSnapshotProvider for FixedWorkloadProvider { + fn workload_admission_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot { + self.snapshot.clone() + } + } + + fn install_provider(snapshot: WorkloadAdmissionRegistrySnapshot) { + set_scanner_workload_admission_snapshot_provider(Arc::new(FixedWorkloadProvider { snapshot })); + } + + #[test] + #[serial] + fn foreground_workload_activity_falls_back_to_local_read_activity() { + clear_scanner_workload_admission_snapshot_provider_for_test(); + crate::reset_foreground_read_activity_for_test(); + crate::set_foreground_read_activity(3); + + assert_eq!(foreground_workload_activity(), 3); + + crate::reset_foreground_read_activity_for_test(); + } + + #[test] + #[serial] + fn foreground_workload_activity_uses_shared_provider_counts() { + clear_scanner_workload_admission_snapshot_provider_for_test(); + crate::reset_foreground_read_activity_for_test(); + install_provider(WorkloadAdmissionRegistrySnapshot::new(vec![ + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, AdmissionState::Open).with_counts( + Some(5), + None, + Some(8), + ), + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, AdmissionState::Open).with_counts( + Some(2), + None, + Some(4), + ), + ])); + + assert_eq!(foreground_workload_activity(), 5); + + clear_scanner_workload_admission_snapshot_provider_for_test(); + } + + #[test] + #[serial] + fn foreground_workload_activity_treats_saturation_without_counts_as_pressure() { + clear_scanner_workload_admission_snapshot_provider_for_test(); + crate::reset_foreground_read_activity_for_test(); + install_provider(WorkloadAdmissionRegistrySnapshot::new(vec![ + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, AdmissionState::Saturated).with_counts( + None, + None, + Some(7), + ), + ])); + + assert_eq!(foreground_workload_activity(), 7); + + clear_scanner_workload_admission_snapshot_provider_for_test(); + } + + #[test] + #[serial] + fn foreground_workload_activity_treats_zero_limit_saturation_as_pressure() { + clear_scanner_workload_admission_snapshot_provider_for_test(); + crate::reset_foreground_read_activity_for_test(); + install_provider(WorkloadAdmissionRegistrySnapshot::new(vec![ + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, AdmissionState::Saturated).with_counts( + None, + None, + Some(0), + ), + ])); + + assert_eq!(foreground_workload_activity(), 1); + + clear_scanner_workload_admission_snapshot_provider_for_test(); + } +} diff --git a/docs/architecture/README.md b/docs/architecture/README.md index 695bc23ca..ec31cffba 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -42,6 +42,8 @@ Required headings and strings in these files are asserted by `scripts/check_arch | [workload-admission-contracts.md](workload-admission-contracts.md) | adding a workload class or snapshot provider, or consuming admission state from a background job | | [background-controller-contract.md](background-controller-contract.md) | adding a status snapshot or reconcile surface for a background service, or being tempted to fold several services into a generic controller | | [background-services-inventory.md](background-services-inventory.md) | you need one audited background service's desired source, current-status inputs, status surface, and declared side effects | +| [scanner-usage-publication.md](scanner-usage-publication.md) | changing scanner data-usage cache publication, quota-visible usage snapshots, scanner cycle recovery, or the persisted scanner usage artifacts | +| [scanner-usage-authority-decision.md](scanner-usage-authority-decision.md) | deciding whether quota admission depends on scanner data usage, removing scanner publication layers, or designing a scanner storage boundary | | [config-model-boundary-adr.md](config-model-boundary-adr.md) | touching the server-config model (`Config`, `KV`, `KVS`) or its persistence, or asking which crate owns which part of server configuration | | [admin-route-action-snapshot.md](admin-route-action-snapshot.md) | adding, moving, or re-authorizing an admin route and needing to know where the route → handler → `AdminAction` contract is enforced | | [kms-bulk-rekey-contract.md](kms-bulk-rekey-contract.md) | changing the bulk envelope re-wrap sweep, its admin endpoints, the re-wrap primitive, or which objects a rekey may touch | diff --git a/docs/architecture/crate-boundaries.md b/docs/architecture/crate-boundaries.md index 542c7582f..e4b188a3e 100644 --- a/docs/architecture/crate-boundaries.md +++ b/docs/architecture/crate-boundaries.md @@ -49,6 +49,20 @@ Outer crates reach ECStore only through `rustfs_ecstore::api`, and only from one - RustFS startup internals are crate-private: only `startup_entrypoint` is a public startup module of the `rustfs` library (`rustfs/src/lib.rs`), and items inside the other `startup_*` modules use crate visibility. - The observability dependency baseline is [obs-ecstore-dependency-inventory.md](obs-ecstore-dependency-inventory.md); observability extraction updates it together with the guard. +## Scanner, Heal, And ECStore + +Heal is split by responsibility, not by the shared word "heal". ECStore owns erasure-set repair primitives: quorum metadata arbitration, EC reconstruction, per-disk rename commit, dangling metadata classification, and orphan data-dir reclamation. These stay in ECStore because they share the same object namespace locks, rename commit model, and data-dir cleanup rules as PUT, DELETE, multipart, lifecycle expiry, rebalance, and decommission. Moving those primitives out would split the lock and commit model across crates. + +`crates/heal` owns repair orchestration: queueing, deduplication, admission, scheduling, resume, MRF replay, replacement-disk tracking, and the admin-facing status/control surface. It reaches storage through `HealStorageAPI`; ECStore-originated repair requests flow back through typed repair channels rather than a Cargo dependency on the heal crate. + +`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. + +`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. + +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. + ## Loss-Prevention Coverage The guard pins specific public re-export lines (its `require_source_line` entries) so contract surfaces cannot silently disappear during cleanup. The canonical lists are the guard script and the owning files, not this page: diff --git a/docs/architecture/heal-concurrency-model.md b/docs/architecture/heal-concurrency-model.md index a71dcca37..3c0b138f5 100644 --- a/docs/architecture/heal-concurrency-model.md +++ b/docs/architecture/heal-concurrency-model.md @@ -3,6 +3,8 @@ **Use this when:** changing heal, PUT/multipart commit, delete, lifecycle expiry, or data-movement code that touches the same `(bucket, object)` commit surface; or evaluating whether RustFS needs a persistent per-object healing marker like MinIO's `x-minio-healing`. **Source of truth:** `crates/ecstore/src/set_disk/ops/heal.rs` (`heal_object_with_explicit_version_regen`, `HealObjectLockKind`, `HEAL_RENAME_INCOMPLETE`), `crates/ecstore/src/set_disk/ops/object.rs` (PUT/DELETE lock sections, `reconcile_old_data_cleanup_receipts`), `crates/ecstore/src/set_disk/core/io_primitives.rs` (`commit_rename_data_dir`, `report_old_data_dir_cleanup`, `reclaim_orphan_data_dirs`), `crates/filemeta/src/fileinfo.rs` (`FileInfo::set_healing`), `crates/heal/src/heal/manager/queue.rs` (dedup keys). +For crate ownership, read [crate-boundaries.md](crate-boundaries.md): ECStore owns erasure-set repair primitives that share this lock and commit model, while `crates/heal` owns repair orchestration. + ## Model Heal and every foreground or background write path serialize on the same object-level namespace write lock (a quorum lock RPC in distributed mode, the in-process lock manager on a single node; granularity is the object, the version component is always `None`), and heal holds its guard across the whole rename commit. MinIO's `x-minio-healing` marker is an out-of-lock defence against version-cleanup logic inside `RenameData` interleaving with a heal commit; RustFS's commit model has no such interleaving, so no persistent marker exists (`x-minio-healing` does not occur in `crates/` or `rustfs/`) and none is needed. Three layers replace it: diff --git a/docs/architecture/scanner-usage-authority-decision.md b/docs/architecture/scanner-usage-authority-decision.md new file mode 100644 index 000000000..db8714b7e --- /dev/null +++ b/docs/architecture/scanner-usage-authority-decision.md @@ -0,0 +1,24 @@ +# Scanner Usage Authority Decision + +**Use this when:** deciding whether quota admission depends on scanner data usage, removing scanner publication layers, or designing a scanner storage boundary. +**Source of truth:** [scanner-usage-publication.md](scanner-usage-publication.md), `crates/ecstore/src/bucket/quota/checker.rs`, and the scanner publication state under `crates/scanner/src/scanner/`. + +## Decision + +Date: 2026-09-03 + +RustFS keeps scanner data usage as authoritative for quota admission. + +This selects option A from the backlog decision record: the scanner publication protocol remains necessary while quota admission consumes scanner usage. The cycle epoch, publication CAS, data-movement fence, tier-registry fence, observed snapshot layer, and persisted usage floor are retained and documented as protocol invariants rather than treated as removable compatibility clutter. + +## Rationale + +Quota is a write-path admission decision, so serving quota from best-effort scanner data would turn temporary scanner lag into under-enforcement. The current design therefore needs an availability story for authoritative usage instead of deleting the proof layers that make it authoritative. + +The scanner usage floor provides that availability story. It is a lower bound used when a complete authoritative snapshot is not available, including cold startup, upgrade recovery, and incomplete-cycle repair. Observed snapshots remain useful for admin and observability, but they do not become quota authority. + +## Consequences + +#2214 is the hard design input for future usage-publication changes. A future proposal may still choose soft quota and MinIO-style best-effort usage, but that would be a product change with its own staged compatibility plan for persisted artifacts. + +#2219 may design the scanner storage boundary against the current authoritative protocol. The interface must include the CAS key-value, cycle lock, usage-floor, observed-snapshot, and recovery-marker capabilities needed by [scanner-usage-publication.md](scanner-usage-publication.md); it must not hide those proof obligations behind a generic object-store trait. diff --git a/docs/architecture/scanner-usage-publication.md b/docs/architecture/scanner-usage-publication.md new file mode 100644 index 000000000..4b4791b1f --- /dev/null +++ b/docs/architecture/scanner-usage-publication.md @@ -0,0 +1,129 @@ +# Scanner Usage Publication Contract + +**Use this when:** changing scanner data-usage persistence, quota-visible usage snapshots, scanner cycle state, dirty-usage catch-up, or the conditions under which an observed scanner snapshot may be served. +**Source of truth:** `crates/scanner/src/scanner/usage_store.rs`, `crates/scanner/src/scanner/cycle_state.rs`, `crates/scanner/src/scanner/backlog.rs`, `crates/scanner/src/scanner/leadership.rs`, `crates/scanner/src/data_usage_define.rs`, and quota fallback behavior in `crates/ecstore/src/bucket/quota/checker.rs`. + +## Ownership Model + +One scanner cycle owns an authoritative publication only after it holds the +cluster scanner leadership claim and proves that the storage publication epoch +has not moved. The leadership claim is persisted in the scanner cycle state, +while data-usage publication admission is owned by ECStore because it knows +whether rebalance, decommission, or another data-movement operation has changed +the generation that scanner results are allowed to describe. + +The scanner may compute usage without publication ownership, but it must not +turn that result into authoritative quota-visible state. A complete publication +therefore has three identities: + +- the scanner leader epoch that owns the cycle; +- the storage publication epoch that fences data movement; +- the per-object CAS revision on the usage object being replaced. + +If any identity changes before commit, the result is a candidate for retry or +observation, not an authoritative baseline. + +## Fences + +The protocol uses separate fences because they exclude different stale inputs. +They must not be collapsed unless the replacement proves the same exclusions. + +| Fence | Owner | Excludes | +|---|---|---| +| Scanner leadership claim | scanner | competing scanner leaders and stale cycle writers | +| Storage publication epoch | ECStore | usage computed across rebalance, decommission, or other data-movement generations | +| Publication lease | scanner peers through ECStore-facing activity probes | remote dirty-usage or maintenance state that has not acknowledged the candidate | +| CAS revision | backing config object store | lost updates to `.usage.v2.json`, `.usage.json`, or cycle-state objects | +| Per-set freshness | scanner aggregation | a merged usage snapshot that combines stale and current set results | +| Tier registry generation | scanner tier accounting | bytes classified against a different warm-tier registry | +| Usage floor identity | scanner publication and ECStore quota fallback | empty or legacy values becoming plausible authoritative quota input | + +A reader that cannot prove the required fence for its surface must fail closed +or use the documented observed path below. It must not synthesize an empty usage +snapshot for a missing or corrupt authoritative object. + +## Persisted Objects + +The persisted objects are part of the compatibility contract. Removing one +requires a compatibility window and a dedicated cleanup entry. + +| Object | Owner | Lifecycle | +|---|---|---| +| `.usage-cache.bin` under each bucket and set | scanner disk walk | Rebuilt by scanner from object metadata. Missing data causes a rescan for that bucket/set; corrupt data is not a complete baseline. | +| `.bloomcycle.bin` | scanner cycle state | CAS-updated by the leader. Missing state starts from an uninitialized cycle; corrupt or future state is quarantined before automatic retry. | +| `.usage.v2.json` and `.usage.json` | scanner authoritative publication | `.usage.v2.json` is the primary complete usage snapshot. `.usage.json` is read only as a legacy or companion baseline when it carries a valid persisted identity. Neither bypasses the v2 epoch fence, and readers may treat a snapshot as authoritative only when its baseline identity and completion fields validate. | +| `.usage.observed.json` | scanner observation path | Written when an authoritative publication cannot be proven but a diagnostic snapshot is still useful. It is never a hard-quota authority. | +| `bucket-metadata/.usage.json` | scanner usage floor | Carries the persisted per-bucket floor used by quota during a degraded authoritative-usage window. It is static until the next complete scanner publication. | +| `.bloomcycle.bin.recovery-required.json` | scanner cycle recovery | Quarantines invalid cycle state with retry evidence. Only scanner recovery code updates or clears it. | +| `.scanner-cycle.lock` | scanner runtime lock | Serializes cycle-level work. A missing lock object is not itself usage evidence. | +| `.scanner-pause-backlog.json` | scanner pause and catch-up ledger | Tracks dirty usage, discovered lifecycle work, and full-scan catch-up while authoritative publication is fenced by data movement. It never grants publication admission. | + +## Observed Snapshots + +Observed snapshots are a diagnostic and availability layer. They may be served +only when the snapshot explicitly reports that it is partial or observational, +and only to consumers that do not make hard quota, durability, or deletion +decisions from it. Admin usage views may expose this state with completeness +flags so operators can see progress while the authoritative publication is +blocked. Quota enforcement must not use an observed snapshot as the current +usage authority. + +When an authoritative snapshot is unavailable, quota admission may use the +persisted usage floor. That is an availability fallback, not a fresh count: live +writes do not advance the floor, and overrun is bounded only by writes accepted +before the next complete scanner publication. If no valid persisted floor is +available, quota remains unavailable and fails closed. + +## Availability Decision + +Decision date: 2026-09-03. + +RustFS keeps scanner usage as the authority for hard quota admission. The +publication protocol therefore remains necessary: leadership, storage epoch, +lease, CAS, observed snapshot, and usage-floor layers are the proof machinery +that lets a distributed background scan feed a quota decision without accepting +stale or cross-generation usage as current truth. + +The availability contract is: + +- the authoritative fast path reads complete in-memory or persisted scanner + usage; +- during upgrade or publication outage, quota may admit against the persisted + per-bucket usage floor; +- the floor is advisory for the outage window and must converge back to a + complete scanner publication; +- a bucket with neither authoritative usage nor a valid floor fails closed. + +Changing this decision to a soft-quota model would be a product change, not a +scanner refactor. It would need a staged removal of the authority-specific +layers and compatibility handling for the persisted objects above. + +## Deletion And Recovery Rules + +Only the owner of an object may delete or quarantine it: + +- scanner may rebuild per-set `.usage-cache.bin` after a scan proves the + replacement contents; +- scanner cycle recovery may quarantine invalid `.bloomcycle.bin` and clear the + marker only after a valid cycle state is persisted; +- scanner publication may replace `.usage.v2.json` or legacy companions only + through the publication fences above; +- quota consumers may read the usage floor but must not delete or repair it; +- operators may reset scanner usage state only through the supported scanner + reset surface, which records the reset paths and forces a full rebuild. + +Missing, undecodable, or identity-less data is not converted to zero. It is +reported as uninitialized, recovery-required, observed-only, or unavailable +according to the reader's surface. + +## Existing Fixes As Invariants + +Several prior scanner fixes are consequences of this contract rather than +standalone patches: + +- incomplete scanner usage must not become a complete admin or quota baseline, + because completeness and floor identity are part of publication ownership; +- dirty usage and maintenance acknowledgements must fence publication, because + a remote node with unacknowledged work can invalidate the candidate; +- a legacy or backup usage object may help recover availability only when it + carries a valid baseline identity and does not cross the primary epoch fence. diff --git a/docs/architecture/workload-admission-contracts.md b/docs/architecture/workload-admission-contracts.md index a1bea9ea3..3d3295e00 100644 --- a/docs/architecture/workload-admission-contracts.md +++ b/docs/architecture/workload-admission-contracts.md @@ -12,9 +12,9 @@ | Class | Provider (`impl WorkloadAdmissionSnapshotProvider`) | `active` / `queued` / `limit` source | Reports `Unknown` when | |---|---|---|---| | `ForegroundRead` | `ConcurrencyManager` in `rustfs/src/storage/concurrency/manager.rs` (source of truth); re-exposed unchanged by the RustFS runtime provider | disk-read permits in use / `None` (the semaphore exposes no waiter count) / configured max concurrent disk reads | the storage registry has no entry | -| `ForegroundWrite` | none | none | always: no write-specific admission owner exposes a read-only surface yet | +| `ForegroundWrite` | `ConcurrencyManager` in `rustfs/src/storage/concurrency/manager.rs` (source of truth); re-exposed unchanged by the RustFS runtime provider | foreground-write permits in use or legacy active-write counter / `None` / configured or derived write-admission limit | the storage registry has no entry | | `Metadata` | `RustFsWorkloadAdmissionSnapshotProvider` in `rustfs/src/workload_admission.rs` | `Open` once the bucket metadata runtime handle exists; no counts | bucket metadata runtime not initialized | -| `Scanner` | same | scanner active work-unit counter / none / none | the counter is zero (idle and uninitialized are indistinguishable) | +| `Scanner` | same | scanner active work-unit counter / none / configured set-scan limit when nonzero | scanner runtime not initialized | | `Repair` | same | heal active tasks / heal queue length / `None` (limits live behind the async heal manager state) | heal manager not initialized | | `Replication` | same | active regular + large-object + MRF workers / site replication queue count / `None` (limits owned by the async pool and resize policy) | replication runtime not initialized, or queue stats currently locked | @@ -28,6 +28,7 @@ Consumers that read the snapshot to self-throttle exist, and they do not change |---|---|---| | Data-movement backpressure (decommission, rebalance) | `crates/ecstore/src/data_movement/backpressure.rs` (`wait_for_data_movement_admission`, `foreground_pressure`) | Delays the next data-movement step while `ForegroundRead` or `ForegroundWrite` usage exceeds the configured high-water percent. ECStore receives the provider through `set_workload_admission_snapshot_provider` (`crates/ecstore/src/lib.rs`), published from `rustfs/src/startup_background.rs`; with no provider the step is admitted immediately. | | Heal manager mainline throttle | `crates/heal/src/heal/manager.rs` (`new_with_workload_provider`) | When `mainline_throttle_enable` is set, defers heal work while `ForegroundRead` or `ForegroundWrite` utilization exceeds the configured high-water percents; with no provider or the throttle disabled, heal pacing is unchanged. | +| Scanner sleeper and scan fan-out | `crates/scanner/src/workload_admission.rs`, `crates/scanner/src/sleeper.rs`, and `crates/scanner/src/scanner_io/guards.rs` | Reads the same provider published from `rustfs/src/startup_background.rs` and combines it with scanner-local foreground read guards. Foreground activity increases scanner sleeps and reduces set/disk scan fan-out to one; cycle budgets still own object, directory, and duration limits. With no provider, scanner keeps the legacy local foreground-read behavior. | ## Boundary Rules diff --git a/docs/testing/ci-gates.md b/docs/testing/ci-gates.md index c12050a40..117fd1e37 100644 --- a/docs/testing/ci-gates.md +++ b/docs/testing/ci-gates.md @@ -57,6 +57,14 @@ Promotion rule: never promote a report-only lane to required from one green run. e2e filters live in `.config/nextest.toml`; extend a profile instead of adding a second selector. Before a profile runs, `scripts/check_test_wiring.py` compares its listing to the committed digest in `.config/e2e--selection.txt`, so a silent test drop fails closed. +Scanner usage and heal rebuild coverage are intentionally split by risk and +cost. `data_usage_test` runs in the PR `e2e-smoke` lane so changes that affect +authoritative scanner usage publication, quota-visible usage, or admin usage +snapshots get an end-to-end signal before merge review. `heal_erasure_disk_rebuild_test` +runs in `e2e-full` so core erasure heal rebuild regressions are caught no later +than the merge queue or `main` push lane; it also remains in `e2e-nightly` with +the serialized cluster fault-domain suites for scheduled soak signal. + ## Scheduled validation Scheduled lanes never block a PR. Their workflow-local gate fails the run, scheduled failures route to the shared failure-issue action, and `scheduled-validation-freshness.yml` fails when a workflow listed in `.github/scheduled-validations.json` has not run within its `max_age_hours` (a `never_ran_grace_until` entry covers the window before a newly enabled cron's first slot). Cadence is qualitative here; the cron lives in each workflow's `on.schedule`. diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index d34f4d928..6d716d63b 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -227,7 +227,7 @@ rustfs-heal = { workspace = true } rustfs-audit = { workspace = true } rustfs-common = { workspace = true } rustfs-heal-contracts = { workspace = true } -rustfs-scanner-contracts = { workspace = true } +rustfs-scanner-metrics = { workspace = true } rustfs-config = { workspace = true, features = ["notify", "server-config-model"] } rustfs-crypto = { workspace = true } rustfs-credentials = { workspace = true } diff --git a/rustfs/src/admin/handlers/scanner.rs b/rustfs/src/admin/handlers/scanner.rs index 638133479..077323463 100644 --- a/rustfs/src/admin/handlers/scanner.rs +++ b/rustfs/src/admin/handlers/scanner.rs @@ -29,10 +29,8 @@ use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; use rustfs_credentials::Credentials; use rustfs_policy::policy::action::{Action, AdminAction}; #[cfg(test)] -use rustfs_scanner_contracts::metrics::ScannerLifecycleTransitionSnapshot; -use rustfs_scanner_contracts::metrics::{ - ScannerLifecycleExpirySnapshot, ScannerMaintenanceControlSnapshot, ScannerMetricsReport, -}; +use rustfs_scanner_metrics::metrics::ScannerLifecycleTransitionSnapshot; +use rustfs_scanner_metrics::metrics::{ScannerLifecycleExpirySnapshot, ScannerMaintenanceControlSnapshot, ScannerMetricsReport}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 95285291d..de8154d93 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -125,7 +125,7 @@ pub(crate) mod runtime_sources { pub(crate) type DailyAllTierStats = super::DailyAllTierStats; pub(crate) type ECStore = super::ECStore; pub(crate) type NotificationSys = super::NotificationSys; - pub(crate) type ScannerMetricsReport = rustfs_scanner_contracts::metrics::ScannerMetricsReport; + pub(crate) type ScannerMetricsReport = rustfs_scanner_metrics::metrics::ScannerMetricsReport; pub(crate) type StorageClassConfig = crate::storage::storage_api::ecstore_config::storageclass::Config; pub(crate) type TierConfigMgr = crate::storage::storage_api::TierConfigMgr; } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index d7e6837a2..11a6c0f09 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -148,7 +148,7 @@ pub(crate) mod runtime { pub(crate) type NotificationSys = crate::storage::storage_api::NotificationSys; pub(crate) type ObjectStoreResolver = crate::storage::storage_api::ObjectStoreResolver; pub(crate) type ReplicationStats = crate::storage::storage_api::ReplicationStats; - pub(crate) type ScannerMetricsReport = rustfs_scanner_contracts::metrics::ScannerMetricsReport; + pub(crate) type ScannerMetricsReport = rustfs_scanner_metrics::metrics::ScannerMetricsReport; pub(crate) type StorageClassConfig = crate::storage::storage_api::ecstore_config::storageclass::Config; pub(crate) type TierConfigMgr = crate::storage::storage_api::TierConfigMgr; pub(crate) type TransitionState = crate::storage::storage_api::TransitionState; @@ -224,7 +224,7 @@ pub(crate) mod runtime { } pub(crate) async fn collect_scanner_metrics_report() -> ScannerMetricsReport { - rustfs_scanner_contracts::metrics::global_metrics().report().await + rustfs_scanner_metrics::metrics::global_metrics().report().await } #[cfg(test)] diff --git a/rustfs/src/cluster_snapshot.rs b/rustfs/src/cluster_snapshot.rs index f061fa981..33abbcf54 100644 --- a/rustfs/src/cluster_snapshot.rs +++ b/rustfs/src/cluster_snapshot.rs @@ -25,7 +25,7 @@ use crate::storage_api::cluster::control_plane::{ use crate::workload_admission::workload_admission_registry_snapshot; use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot}; use rustfs_io_metrics::internode_metrics::{InternodeMetricsSnapshot, global_internode_metrics}; -use rustfs_scanner_contracts::metrics::{ScannerMetricsReport, global_metrics}; +use rustfs_scanner_metrics::metrics::{ScannerMetricsReport, global_metrics}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct ClusterReadOnlySnapshot { diff --git a/rustfs/src/startup_background.rs b/rustfs/src/startup_background.rs index 4d2077b0c..01b0886de 100644 --- a/rustfs/src/startup_background.rs +++ b/rustfs/src/startup_background.rs @@ -59,6 +59,7 @@ pub(crate) async fn init_background_service_runtime(store: Arc) -> Resu let workload_provider: Arc = Arc::new(RustFsWorkloadAdmissionSnapshotProvider); let _ = set_workload_admission_snapshot_provider(workload_provider.clone()); + rustfs_scanner::set_scanner_workload_admission_snapshot_provider(workload_provider.clone()); if enable_heal || enable_scanner { let heal_storage = Arc::new(ECStoreHealStorage::new(store.clone())); diff --git a/rustfs/src/startup_runtime_sources.rs b/rustfs/src/startup_runtime_sources.rs index ab63777b3..90304d00f 100644 --- a/rustfs/src/startup_runtime_sources.rs +++ b/rustfs/src/startup_runtime_sources.rs @@ -55,7 +55,7 @@ pub(crate) async fn publish_server_addr(addr: &str) { } pub(crate) async fn publish_init_time_now() { - rustfs_scanner_contracts::set_global_init_time_now().await; + rustfs_scanner_metrics::set_global_init_time_now().await; } pub(crate) fn init_kms_service_manager() -> Arc {