gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-30 00:47:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: migrate consumers off rustfs-common heal/scanner shims (#6623)

Browse Source 
* refactor(ecstore): import heal/scanner contracts crates directly (backlog#1843)

* refactor(heal): import heal/scanner contracts crates directly (backlog#1843)

* refactor(lifecycle): import heal/scanner contracts crates directly (backlog#1843)

* refactor(obs): import heal/scanner contracts crates directly (backlog#1843)

* refactor(protos): import heal/scanner contracts crates directly (backlog#1843)

* refactor(scanner): import heal/scanner contracts crates directly (backlog#1843)

* refactor(rustfs): import heal/scanner contracts crates directly (backlog#1843)
This commit is contained in:
Zhengchao An
2026-08-26 21:13:18 +08:00
committed by GitHub
parent 2ada8a5cfb
commit a96dd7d289
72 changed files with 263 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/Cargo.toml
+2
View File
@@ -143,6 +143,8 @@ rustfs-config = { workspace = true, features = ["notify", "audit", "server-confi
rustfs-concurrency.workspace = true
rustfs-credentials = { workspace = true }
rustfs-common.workspace = true
rustfs-heal-contracts.workspace = true
rustfs-scanner-contracts.workspace = true
rustfs-policy.workspace = true
rustfs-protos.workspace = true
rustfs-replication.workspace = true
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_audit.rs
+1 -1
View File
@@ -16,8 +16,8 @@ use super::runtime_boundary as runtime_sources;
use crate::bucket::lifecycle::lifecycle;
use crate::object_api::ObjectInfo;
use crate::services::event_notification::{EventArgs, send_event};
use rustfs_common::metrics::IlmAction;
use rustfs_s3_types::EventName;
use rustfs_scanner_contracts::metrics::IlmAction;
const LIFECYCLE_EXPIRY_USER_AGENT: &str = "Internal: [ILM-Expiry]";
const LIFECYCLE_TRANSITION_USER_AGENT: &str = "Internal: [ILM-Transition]";
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+4 -4
View File
@@ -72,9 +72,6 @@ use crate::store::ECStore;
use async_channel::{Receiver as A_Receiver, Sender as A_Sender, bounded};
use http::HeaderMap;
use rand::RngExt as _;
use rustfs_common::metrics::{
IlmAction, Metrics, ScannerLifecycleExpiryStateUpdate, ScannerLifecycleTransitionStateUpdate, global_metrics,
};
use rustfs_config::{
DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_QUEUE_SEND_TIMEOUT_MS, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX,
DEFAULT_TRANSITION_WORKERS_CAP, ENV_MAX_EXPIRY_WORKERS, ENV_TRANSITION_QUEUE_CAPACITY, ENV_TRANSITION_QUEUE_SEND_TIMEOUT_MS,
@@ -84,6 +81,9 @@ 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::{
IlmAction, Metrics, ScannerLifecycleExpiryStateUpdate, ScannerLifecycleTransitionStateUpdate, global_metrics,
};
use rustfs_utils::{
get_env_i64, get_env_usize,
path::encode_dir_object,
@@ -5460,11 +5460,11 @@ mod tests {
use futures::FutureExt;
#[cfg(feature = "test-util")]
use http::HeaderMap;
use rustfs_common::metrics::{IlmAction, global_metrics};
use rustfs_config::ENV_MAX_EXPIRY_WORKERS;
use rustfs_config::ENV_TRANSITION_WORKERS_ABSOLUTE_MAX;
use rustfs_data_usage::TierStats;
use rustfs_filemeta::{FileInfo, FileMeta};
use rustfs_scanner_contracts::metrics::{IlmAction, global_metrics};
use s3s::dto::{
BucketLifecycleConfiguration, DefaultRetention, ExpirationStatus, LifecycleExpiration, LifecycleRule, MetadataEntry,
ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRetentionMode, ObjectLockRule, OutputLocation, RestoreRequest,
crates/ecstore/src/bucket/lifecycle/replication_sink.rs
+2 -2
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use rustfs_common::metrics::IlmAction;
use rustfs_scanner_contracts::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_common::metrics::IlmAction;
use rustfs_scanner_contracts::metrics::IlmAction;
use s3s::dto::{
BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication,
DeleteReplicationStatus, Destination, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus,
crates/ecstore/src/bucket/metadata_sys.rs
+1 -1
View File
@@ -27,7 +27,7 @@ use crate::storage_api_contracts::heal::HealOperations as _;
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
use crate::store::{ECStore, await_bucket_namespace_operation};
use futures::future::join_all;
use rustfs_common::heal_channel::HealOpts;
use rustfs_heal_contracts::heal_channel::HealOpts;
use rustfs_policy::policy::BucketPolicy;
use s3s::dto::ReplicationConfiguration;
use s3s::dto::{
crates/ecstore/src/bucket/quota/checker.rs
+5 -5
View File
@@ -15,8 +15,8 @@
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_common::metrics::Metric;
use rustfs_config::QUOTA_CONFIG_FILE;
use rustfs_scanner_contracts::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_common::metrics::Metrics::inc_time(Metric::QuotaCheck, duration);
rustfs_scanner_contracts::metrics::Metrics::inc_time(Metric::QuotaCheck, duration);
if !allowed {
rustfs_common::metrics::Metrics::inc_time(Metric::QuotaViolation, duration);
rustfs_scanner_contracts::metrics::Metrics::inc_time(Metric::QuotaViolation, duration);
}
Ok(result)
@@ -185,7 +185,7 @@ impl QuotaChecker {
.await
.map_err(QuotaError::StorageError)?;
rustfs_common::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed());
rustfs_scanner_contracts::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed());
Ok(updated_at)
}
@@ -206,7 +206,7 @@ impl QuotaChecker {
}
.map_err(QuotaError::StorageError)?;
rustfs_common::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed());
rustfs_scanner_contracts::metrics::Metrics::inc_time(Metric::QuotaSync, start_time.elapsed());
Ok(updated_at)
}
crates/ecstore/src/cluster/rpc/peer_s3_client.rs
+1 -1
View File
@@ -36,7 +36,7 @@ use crate::{
};
use async_trait::async_trait;
use futures::future::join_all;
use rustfs_common::heal_channel::{DriveState, HealItemType, HealOpts, RUSTFS_RESERVED_BUCKET};
use rustfs_heal_contracts::heal_channel::{DriveState, HealItemType, HealOpts, RUSTFS_RESERVED_BUCKET};
use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem};
use rustfs_protos::proto_gen::node_service::node_service_client::NodeServiceClient;
use rustfs_protos::proto_gen::node_service::{
crates/ecstore/src/config/com.rs
+2 -2
View File
@@ -27,7 +27,6 @@ use crate::storage_api_contracts::{
};
use crate::store::ECStore;
use http::HeaderMap;
use rustfs_common::heal_channel::{HealOpts, HealScanMode};
use rustfs_config::audit::{
AUDIT_AMQP_KEYS, AUDIT_AMQP_SUB_SYS, AUDIT_KAFKA_KEYS, AUDIT_KAFKA_SUB_SYS, AUDIT_MQTT_KEYS, AUDIT_MQTT_SUB_SYS,
AUDIT_MYSQL_KEYS, AUDIT_MYSQL_SUB_SYS, AUDIT_NATS_KEYS, AUDIT_NATS_SUB_SYS, AUDIT_POSTGRES_KEYS, AUDIT_POSTGRES_SUB_SYS,
@@ -46,6 +45,7 @@ use rustfs_config::{
SCANNER_SUB_SYS,
};
use rustfs_filemeta::FileInfo;
use rustfs_heal_contracts::heal_channel::{HealOpts, HealScanMode};
use serde_json::{Map, Value};
use std::collections::{HashMap, HashSet};
use std::sync::LazyLock;
@@ -4710,7 +4710,7 @@ mod tests {
decode_persisted_server_config, fallback_server_config_after_corruption, is_server_config_corrupt_error,
read_config_without_migrate_with_recovery, replace_server_config_decrypt_fn_for_test,
};
use rustfs_common::heal_channel::HealOpts;
use rustfs_heal_contracts::heal_channel::HealOpts;
use std::sync::Mutex;
/// Bytes mirroring issue #4156: a bitrot-corrupted `config.json` whose
crates/ecstore/src/core/pools.rs
+1 -1
View File
@@ -70,8 +70,8 @@ use http::HeaderMap;
#[cfg(test)]
use rmp_serde::Deserializer;
use rmp_serde::Serializer;
use rustfs_common::heal_channel::HealOpts;
use rustfs_filemeta::{FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
use rustfs_heal_contracts::heal_channel::HealOpts;
use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum};
use rustfs_utils::path::{encode_dir_object, path_join, path_to_bucket_object, path_to_bucket_object_with_base_path};
use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, ReplicationConfiguration};
crates/ecstore/src/core/sets.rs
+2 -2
View File
@@ -46,9 +46,9 @@ use futures::{
stream::{FuturesUnordered, StreamExt},
};
use http::HeaderMap;
use rustfs_common::heal_channel::HealOpts;
use rustfs_common::heal_channel::{DriveState, HealItemType};
use rustfs_filemeta::FileInfo;
use rustfs_heal_contracts::heal_channel::HealOpts;
use rustfs_heal_contracts::heal_channel::{DriveState, HealItemType};
use rustfs_lock::NamespaceLockWrapper;
use rustfs_madmin::heal_commands::HealResultItem;
use rustfs_utils::{crc_hash, path::path_join_buf, sip_hash};
crates/ecstore/src/diagnostics/admin_server_info.rs
+1 -1
View File
@@ -26,7 +26,7 @@ use crate::{
use crate::data_usage::load_data_usage_cache;
use crate::storage_api_contracts::admin::StorageAdminApi;
use crate::storage_api_contracts::bucket::BucketOptions;
use rustfs_common::heal_channel::DriveState;
use rustfs_heal_contracts::heal_channel::DriveState;
use rustfs_madmin::{
BackendDisks, Disk, ErasureSetInfo, ITEM_INITIALIZING, ITEM_OFFLINE, ITEM_ONLINE, ITEM_UNKNOWN, InfoMessage, MemStats,
ServerProperties,
crates/ecstore/src/ecstore_validation_blackbox.rs
+2 -2
View File
@@ -208,7 +208,7 @@ async fn blackbox_get_restores_body_after_one_shard_file_is_removed() {
// Serialized: forces the reader-setup strategy through a process-global env var.
#[serial_test::serial]
async fn blackbox_heal_requests_preserve_repair_scope() {
use rustfs_common::heal_channel::{
use rustfs_heal_contracts::heal_channel::{
HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest, HealRequestSource,
};
@@ -238,7 +238,7 @@ async fn blackbox_heal_requests_preserve_repair_scope() {
// (failing their submitter, which releases their dedup reservation), and
// fail fast once the receiver drops at test end. Tests that must observe a
// deterministic channel state serialize under the same serial key.
let mut heal_rx = rustfs_common::heal_channel::init_heal_channel()
let mut heal_rx = rustfs_heal_contracts::heal_channel::init_heal_channel()
.expect("this must be the only ecstore test that owns the heal channel receiver");
// Ordinary PUTs use the same admission channel as read repair. A single
crates/ecstore/src/layout/set_heal.rs
+1 -1
View File
@@ -14,7 +14,7 @@
use crate::disk::{DiskInfo, error::DiskError};
use crate::layout::{endpoints::Endpoints, format::FormatV3};
use rustfs_common::heal_channel::DriveState;
use rustfs_heal_contracts::heal_channel::DriveState;
use rustfs_madmin::heal_commands::HealDriveInfo;
pub(crate) fn formats_to_drives_info(
crates/ecstore/src/object_api/types.rs
+1 -1
View File
@@ -281,7 +281,7 @@ pub struct QuotaAdmission {
pub struct LifecycleDeleteAllRequest {
pub(crate) version_id: Option<Uuid>,
pub(crate) delete_marker: bool,
pub(crate) action: rustfs_common::metrics::IlmAction,
pub(crate) action: rustfs_scanner_contracts::metrics::IlmAction,
pub(crate) rule_id: String,
pub(crate) phase: LifecycleDeleteAllPhase,
}
crates/ecstore/src/services/metrics_realtime.rs
+27 -26
View File
@@ -18,7 +18,7 @@ use crate::storage_api_contracts::admin::StorageAdminApi;
#[cfg(test)]
use chrono::Utc;
use jiff::Timestamp;
use rustfs_common::{heal_channel::DriveState, metrics::global_metrics};
use rustfs_heal_contracts::heal_channel::DriveState;
use rustfs_io_metrics::internode_metrics::global_internode_metrics;
use rustfs_madmin::metrics::{
DiskIOStats, DiskMetric, LastMinute as MadminLastMinute, NetDevLine, NetMetrics, RPCMetrics, RealtimeMetrics,
@@ -32,6 +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_utils::os::get_drive_stats;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
@@ -81,7 +82,7 @@ fn unix_millis_to_jiff_timestamp(millis: u64, fallback: Timestamp) -> Timestamp
}
}
fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsReport) -> MadminScannerMetrics {
fn to_madmin_scanner_metrics(metrics: rustfs_scanner_contracts::metrics::ScannerMetricsReport) -> MadminScannerMetrics {
MadminScannerMetrics {
collected_at: metrics.collected_at,
current_cycle: metrics.current_cycle,
@@ -563,8 +564,8 @@ async fn collect_local_disks_metrics(disks: &HashSet<String>) -> HashMap<String,
#[cfg(test)]
mod test {
use super::*;
use rustfs_common::metrics::CurrentCycle;
use rustfs_io_metrics::internode_metrics::global_internode_metrics;
use rustfs_scanner_contracts::metrics::CurrentCycle;
use serial_test::serial;
use std::time::Duration;
@@ -618,7 +619,7 @@ mod test {
#[test]
fn scanner_metrics_mapping_preserves_partial_source_status() {
let current_started = Utc::now() - chrono::Duration::seconds(5);
let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport {
let scanner = to_madmin_scanner_metrics(rustfs_scanner_contracts::metrics::ScannerMetricsReport {
current_cycle_active: true,
current_started: chrono_to_jiff_timestamp(current_started),
last_cycle_partial_source: "usage".to_string(),
@@ -627,7 +628,7 @@ mod test {
cycle_recovery_required_total: 2,
cycle_last_progress_age: 17,
leader_lease_without_progress: true,
partial_cycles_by_source: vec![rustfs_common::metrics::ScannerSourceCycleSnapshot {
partial_cycles_by_source: vec![rustfs_scanner_contracts::metrics::ScannerSourceCycleSnapshot {
source: "usage".to_string(),
cycles: 2,
}],
@@ -653,11 +654,11 @@ mod test {
#[tokio::test]
#[serial]
async fn collect_local_metrics_preserves_scanner_cycle_started_time() {
let previous_init_time = *rustfs_common::globals::GLOBAL_INIT_TIME.read().await;
let previous_init_time = *rustfs_scanner_contracts::GLOBAL_INIT_TIME.read().await;
let previous_cycle = global_metrics().get_cycle().await;
let init_time = Utc::now() - chrono::Duration::hours(1);
let cycle_started = Utc::now() - chrono::Duration::seconds(5);
*rustfs_common::globals::GLOBAL_INIT_TIME.write().await = Some(init_time);
*rustfs_scanner_contracts::GLOBAL_INIT_TIME.write().await = Some(init_time);
let cycle = CurrentCycle {
current: 0,
next: 1,
@@ -672,7 +673,7 @@ mod test {
.finish_scan_cycle_work_with_cycle(cycle_start, previous_cycle.clone().unwrap_or_default())
.await;
global_metrics().set_cycle(previous_cycle).await;
*rustfs_common::globals::GLOBAL_INIT_TIME.write().await = previous_init_time;
*rustfs_scanner_contracts::GLOBAL_INIT_TIME.write().await = previous_init_time;
let encoded = rmp_serde::to_vec_named(&realtime).expect("realtime metrics should encode");
let decoded: RealtimeMetrics = rmp_serde::from_slice(&encoded).expect("realtime metrics should decode");
@@ -685,8 +686,8 @@ mod test {
#[test]
fn scanner_metrics_mapping_preserves_pacing_pressure() {
let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport {
pacing_pressure: rustfs_common::metrics::ScannerPacingPressureSnapshot {
let scanner = to_madmin_scanner_metrics(rustfs_scanner_contracts::metrics::ScannerMetricsReport {
pacing_pressure: rustfs_scanner_contracts::metrics::ScannerPacingPressureSnapshot {
primary_pressure: "cycle_budget".to_string(),
current_queued_scans: 4,
current_active_scans: 2,
@@ -711,12 +712,12 @@ mod test {
#[test]
fn scanner_metrics_mapping_preserves_lifecycle_transition_status() {
let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport {
let scanner = to_madmin_scanner_metrics(rustfs_scanner_contracts::metrics::ScannerMetricsReport {
current_cycle_lifecycle_expiry_actions: 2,
current_cycle_lifecycle_transition_actions: 3,
last_cycle_lifecycle_expiry_actions: 5,
last_cycle_lifecycle_transition_actions: 7,
lifecycle_expiry: rustfs_common::metrics::ScannerLifecycleExpirySnapshot {
lifecycle_expiry: rustfs_scanner_contracts::metrics::ScannerLifecycleExpirySnapshot {
current_queue_capacity: 16,
current_queued: 5,
current_active: 2,
@@ -728,7 +729,7 @@ mod test {
scanner_not_enqueued: 2,
delete_failed: 1,
},
lifecycle_transition: rustfs_common::metrics::ScannerLifecycleTransitionSnapshot {
lifecycle_transition: rustfs_scanner_contracts::metrics::ScannerLifecycleTransitionSnapshot {
current_queue_capacity: 16,
current_queued: 5,
current_active: 2,
@@ -777,10 +778,10 @@ mod test {
#[test]
fn scanner_metrics_mapping_preserves_maintenance_control_status() {
let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport {
maintenance_control: rustfs_common::metrics::ScannerMaintenanceControlSnapshot {
let scanner = to_madmin_scanner_metrics(rustfs_scanner_contracts::metrics::ScannerMetricsReport {
maintenance_control: rustfs_scanner_contracts::metrics::ScannerMaintenanceControlSnapshot {
primary_control: "blocked_source".to_string(),
sources: vec![rustfs_common::metrics::ScannerMaintenanceSourceSnapshot {
sources: vec![rustfs_scanner_contracts::metrics::ScannerMaintenanceSourceSnapshot {
source: "lifecycle".to_string(),
state: "blocked".to_string(),
reason: "missed_work".to_string(),
@@ -814,8 +815,8 @@ mod test {
#[test]
fn scanner_metrics_mapping_preserves_usage_freshness_status() {
let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport {
usage_freshness: rustfs_common::metrics::ScannerUsageFreshnessSnapshot {
let scanner = to_madmin_scanner_metrics(rustfs_scanner_contracts::metrics::ScannerMetricsReport {
usage_freshness: rustfs_scanner_contracts::metrics::ScannerUsageFreshnessSnapshot {
dirty_pending_buckets: 3,
last_dirty_mark_unix_secs: 10,
last_dirty_clear_unix_secs: 11,
@@ -856,7 +857,7 @@ mod test {
#[test]
fn scanner_metrics_mapping_preserves_distributed_status_fields() {
let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport {
let scanner = to_madmin_scanner_metrics(rustfs_scanner_contracts::metrics::ScannerMetricsReport {
active_scan_paths: 2,
oldest_active_path_age_seconds: 45,
active_paths: vec!["disk-a/bucket-a".to_string(), "disk-b/bucket-b".to_string()],
@@ -912,7 +913,7 @@ mod test {
cycle_max_directories: 38,
bitrot_cycle_enabled: true,
bitrot_cycle_seconds: 39.0,
scan_checkpoint: Some(rustfs_common::metrics::ScannerCheckpointReport {
scan_checkpoint: Some(rustfs_scanner_contracts::metrics::ScannerCheckpointReport {
version: 1,
resume_after: "bucket-a/prefix-a".to_string(),
reason: "directories".to_string(),
@@ -922,7 +923,7 @@ mod test {
scan_checkpoint_cleared: 41,
scan_checkpoint_ignored: 42,
scan_checkpoint_stale: 43,
source_work: vec![rustfs_common::metrics::ScannerSourceWorkSnapshot {
source_work: vec![rustfs_scanner_contracts::metrics::ScannerSourceWorkSnapshot {
source: "usage".to_string(),
checked: 44,
queued: 45,
@@ -931,7 +932,7 @@ mod test {
skipped: 48,
missed: 49,
}],
current_cycle_source_work: vec![rustfs_common::metrics::ScannerSourceWorkSnapshot {
current_cycle_source_work: vec![rustfs_scanner_contracts::metrics::ScannerSourceWorkSnapshot {
source: "lifecycle".to_string(),
checked: 50,
queued: 51,
@@ -940,7 +941,7 @@ mod test {
skipped: 54,
missed: 55,
}],
last_cycle_source_work: vec![rustfs_common::metrics::ScannerSourceWorkSnapshot {
last_cycle_source_work: vec![rustfs_scanner_contracts::metrics::ScannerSourceWorkSnapshot {
source: "heal".to_string(),
checked: 56,
queued: 57,
@@ -949,7 +950,7 @@ mod test {
skipped: 60,
missed: 61,
}],
replication_repair: vec![rustfs_common::metrics::ScannerReplicationRepairSnapshot {
replication_repair: vec![rustfs_scanner_contracts::metrics::ScannerReplicationRepairSnapshot {
source: "bucket_replication".to_string(),
kind: "object".to_string(),
scanner_role: "repair_admission".to_string(),
@@ -961,7 +962,7 @@ mod test {
skipped: 66,
missed: 67,
}],
current_cycle_replication_repair: vec![rustfs_common::metrics::ScannerReplicationRepairSnapshot {
current_cycle_replication_repair: vec![rustfs_scanner_contracts::metrics::ScannerReplicationRepairSnapshot {
source: "bucket_replication".to_string(),
kind: "delete_marker".to_string(),
scanner_role: "repair_admission".to_string(),
@@ -973,7 +974,7 @@ mod test {
skipped: 72,
missed: 73,
}],
last_cycle_replication_repair: vec![rustfs_common::metrics::ScannerReplicationRepairSnapshot {
last_cycle_replication_repair: vec![rustfs_scanner_contracts::metrics::ScannerReplicationRepairSnapshot {
source: "site_replication".to_string(),
kind: "active_resync".to_string(),
scanner_role: "boundary_signal".to_string(),
crates/ecstore/src/services/notification_sys.rs
+1 -1
View File
@@ -2089,7 +2089,7 @@ async fn peer_disk_health(host: &str) -> Option<PeerDiskHealth> {
disks.push(rustfs_madmin::Disk {
endpoint: ep.to_string(),
state: if online {
rustfs_common::heal_channel::DriveState::Ok.to_string()
rustfs_heal_contracts::heal_channel::DriveState::Ok.to_string()
} else {
ItemState::Offline.to_string().to_owned()
},
crates/ecstore/src/set_disk/core/io_primitives.rs
+22 -15
View File
@@ -1364,7 +1364,7 @@ pub(in crate::set_disk) enum ReadRepairAdmissionOutcome {
pub(in crate::set_disk) type ReadRepairAdmissionFuture = Pin<Box<dyn Future<Output = ReadRepairAdmissionOutcome> + Send>>;
pub(in crate::set_disk) type ReadRepairAdmissionSubmitter =
fn(rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture;
fn(rustfs_heal_contracts::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture;
pub(in crate::set_disk) struct ReadRepairHealSubmission<'a> {
pub(in crate::set_disk) bucket: &'a str,
@@ -1385,7 +1385,7 @@ pub(in crate::set_disk) struct ReadRepairHealSubmission<'a> {
}
pub(in crate::set_disk) fn send_read_repair_heal_request(
request: rustfs_common::heal_channel::HealChannelRequest,
request: rustfs_heal_contracts::heal_channel::HealChannelRequest,
) -> ReadRepairAdmissionFuture {
Box::pin(async {
match send_heal_request_with_admission(request).await {
@@ -1453,7 +1453,7 @@ pub(in crate::set_disk) async fn submit_read_repair_heal_with_submitter(
let _ = rustfs_common::mrf_channel::try_send_mrf_intent_typed(kind, bucket, object, version_uuid, Some(scope));
}
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
let mut request = rustfs_heal_contracts::heal_channel::create_heal_request_with_options(
bucket.to_string(),
Some(object.to_string()),
false,
@@ -3580,7 +3580,7 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
tail_drain: tokio::task::JoinHandle<Option<RenameTailOutcome>>,
guard_release: tokio::sync::oneshot::Receiver<bool>,
guards: Guards,
request: rustfs_common::heal_channel::HealChannelRequest,
request: rustfs_heal_contracts::heal_channel::HealChannelRequest,
finalize: Finalize,
cleanup: Cleanup,
submit: Submit,
@@ -3590,7 +3590,7 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
FinalizeFuture: Future<Output = ()> + Send,
Cleanup: FnOnce(Guards, Vec<RenameTailCleanup>) -> CleanupFuture + Send,
CleanupFuture: Future<Output = ()> + Send,
Submit: FnOnce(rustfs_common::heal_channel::HealChannelRequest) -> SubmitFuture + Send,
Submit: FnOnce(rustfs_heal_contracts::heal_channel::HealChannelRequest) -> SubmitFuture + Send,
SubmitFuture: Future<Output = ()> + Send,
{
let (needs_heal, tail_cleanup, tail_complete) = match tail_drain.await {
@@ -4939,16 +4939,17 @@ impl SetDisks {
// reclaim_orphan_data_dirs. Reuses the existing heal channel, which
// deduplicates and back-pressures via admission; failures only drop
// the return value (same shape as multipart's existing heal enqueue).
let _ =
rustfs_common::heal_channel::send_heal_request(rustfs_common::heal_channel::create_heal_request_with_options(
let _ = rustfs_heal_contracts::heal_channel::send_heal_request(
rustfs_heal_contracts::heal_channel::create_heal_request_with_options(
bucket.to_string(),
Some(object.to_string()),
false,
Some(rustfs_common::heal_channel::HealChannelPriority::Normal),
Some(rustfs_heal_contracts::heal_channel::HealChannelPriority::Normal),
Some(self.pool_index),
Some(self.set_index),
))
.await;
),
)
.await;
}
}
@@ -6800,18 +6801,24 @@ mod tests {
write_raw_file_meta_unchecked(disk, bucket, object, metadata).await;
}
fn failed_read_repair_submitter(_request: rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture {
fn failed_read_repair_submitter(
_request: rustfs_heal_contracts::heal_channel::HealChannelRequest,
) -> ReadRepairAdmissionFuture {
Box::pin(async { ReadRepairAdmissionOutcome::Failed("injected submit failure".to_string()) })
}
fn accepted_read_repair_submitter(_request: rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture {
fn accepted_read_repair_submitter(
_request: rustfs_heal_contracts::heal_channel::HealChannelRequest,
) -> ReadRepairAdmissionFuture {
Box::pin(async { ReadRepairAdmissionOutcome::Response(HealAdmissionResult::Accepted) })
}
fn dropped_read_repair_submitter(_request: rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture {
fn dropped_read_repair_submitter(
_request: rustfs_heal_contracts::heal_channel::HealChannelRequest,
) -> ReadRepairAdmissionFuture {
Box::pin(async {
ReadRepairAdmissionOutcome::Response(HealAdmissionResult::Dropped(
rustfs_common::heal_channel::HealAdmissionDropReason::PolicyDropped,
rustfs_heal_contracts::heal_channel::HealAdmissionDropReason::PolicyDropped,
))
})
}
@@ -8853,7 +8860,7 @@ mod tests {
tail_drain,
released,
(),
rustfs_common::heal_channel::HealChannelRequest::default(),
rustfs_heal_contracts::heal_channel::HealChannelRequest::default(),
move || async move {
*finalize_captured.lock().expect("finalize recorder should not poison") = true;
},
crates/ecstore/src/set_disk/mod.rs
+13 -9
View File
@@ -130,15 +130,15 @@ use http::HeaderMap;
use md5::{Digest as Md5Digest, Md5};
use rand::{Rng, seq::SliceRandom};
use regex::Regex;
use rustfs_common::heal_channel::{
DriveState, HealAdmissionResult, HealChannelPriority, HealItemType, HealOpts, HealRequestSource, HealScanMode,
send_heal_disk, send_heal_request_with_admission,
};
use rustfs_config::MI_B;
use rustfs_filemeta::{
FileInfo, FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo,
RawFileInfo, file_info_from_raw, merge_file_meta_versions,
};
use rustfs_heal_contracts::heal_channel::{
DriveState, HealAdmissionResult, HealChannelPriority, HealItemType, HealOpts, HealRequestSource, HealScanMode,
send_heal_disk, send_heal_request_with_admission,
};
use rustfs_io_metrics::{
record_object_lock_diag_acquire_duration, record_object_lock_diag_enabled, record_object_lock_diag_hold_duration,
record_object_lock_diag_slow_acquire, record_object_lock_diag_slow_hold,
@@ -3111,8 +3111,9 @@ pub struct SetDisks {
#[cfg(test)]
storage_class_config_override: Arc<std::sync::RwLock<Option<Arc<storageclass::Config>>>>,
#[cfg(test)]
rename_tail_heal_capture:
Arc<std::sync::Mutex<Option<tokio::sync::mpsc::UnboundedSender<rustfs_common::heal_channel::HealChannelRequest>>>>,
rename_tail_heal_capture: Arc<
std::sync::Mutex<Option<tokio::sync::mpsc::UnboundedSender<rustfs_heal_contracts::heal_channel::HealChannelRequest>>>,
>,
}
// DistributedLock sends the raw ObjectKey to its clients; LockRegistry clones
@@ -3388,7 +3389,10 @@ impl DiskHealthEntry {
}
impl SetDisks {
pub(in crate::set_disk) async fn submit_rename_tail_heal(&self, request: rustfs_common::heal_channel::HealChannelRequest) {
pub(in crate::set_disk) async fn submit_rename_tail_heal(
&self,
request: rustfs_heal_contracts::heal_channel::HealChannelRequest,
) {
#[cfg(test)]
{
let capture = self
@@ -3402,13 +3406,13 @@ impl SetDisks {
}
}
let _ = rustfs_common::heal_channel::send_heal_request(request).await;
let _ = rustfs_heal_contracts::heal_channel::send_heal_request(request).await;
}
#[cfg(test)]
pub(in crate::set_disk) fn capture_test_rename_tail_heals(
&self,
) -> tokio::sync::mpsc::UnboundedReceiver<rustfs_common::heal_channel::HealChannelRequest> {
) -> tokio::sync::mpsc::UnboundedReceiver<rustfs_heal_contracts::heal_channel::HealChannelRequest> {
let (capture, requests) = tokio::sync::mpsc::unbounded_channel();
let mut slot = self
.rename_tail_heal_capture
crates/ecstore/src/set_disk/ops/heal.rs
+1 -1
View File
@@ -2386,8 +2386,8 @@ mod heal_result_report_tests {
store::init_format::{load_format_erasure, save_format_file},
};
use bytes::Bytes;
use rustfs_common::heal_channel::{DriveState, HealOpts, HealScanMode};
use rustfs_filemeta::{BLOCK_SIZE_V2, FileInfo, ObjectPartInfo, TRANSITION_COMPLETE};
use rustfs_heal_contracts::heal_channel::{DriveState, HealOpts, HealScanMode};
use std::sync::{Arc, Mutex};
use tempfile::TempDir;
use time::OffsetDateTime;
crates/ecstore/src/set_disk/ops/multipart.rs
+3 -3
View File
@@ -2749,7 +2749,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
needs_immediate_heal = rename_commit.needs_immediate_heal();
if let Some(rename_tail_drain) = rename_commit.tail_drain.take() {
tail_owns_staging_cleanup = true;
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
let mut request = rustfs_heal_contracts::heal_channel::create_heal_request_with_options(
commit_bucket.clone(),
Some(commit_object.clone()),
false,
@@ -2846,7 +2846,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
let committed_file_info = rename_commit.committed_file_info;
if needs_immediate_heal {
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
let mut request = rustfs_heal_contracts::heal_channel::create_heal_request_with_options(
commit_bucket.clone(),
Some(commit_object.clone()),
false,
@@ -2859,7 +2859,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
.or_else(|| commit_version_suspended.then(Uuid::nil))
.map(|version_id| version_id.to_string());
tokio::spawn(async move {
let _ = rustfs_common::heal_channel::send_heal_request(request).await;
let _ = rustfs_heal_contracts::heal_channel::send_heal_request(request).await;
});
}
crates/ecstore/src/set_disk/ops/object.rs
+8 -8
View File
@@ -352,7 +352,7 @@ mod lifecycle_delete_all_plan_tests {
crate::object_api::LifecycleDeleteAllRequest {
version_id: Some(version_id),
delete_marker: true,
action: rustfs_common::metrics::IlmAction::DelMarkerDeleteAllVersionsAction,
action: rustfs_scanner_contracts::metrics::IlmAction::DelMarkerDeleteAllVersionsAction,
rule_id: "rule".to_string(),
phase: crate::object_api::LifecycleDeleteAllPhase::Preflight,
}
@@ -545,7 +545,7 @@ mod lifecycle_delete_all_plan_tests {
let request = crate::object_api::LifecycleDeleteAllRequest {
version_id: None,
delete_marker: false,
action: rustfs_common::metrics::IlmAction::DeleteAllVersionsAction,
action: rustfs_scanner_contracts::metrics::IlmAction::DeleteAllVersionsAction,
rule_id: "rule".to_string(),
phase: crate::object_api::LifecycleDeleteAllPhase::Preflight,
};
@@ -3242,7 +3242,7 @@ impl SetDisks {
needs_immediate_heal = rename_commit.needs_immediate_heal();
if let Some(rename_tail_drain) = rename_commit.tail_drain.take() {
tail_owns_tmp_cleanup = true;
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
let mut request = rustfs_heal_contracts::heal_channel::create_heal_request_with_options(
commit_bucket.clone(),
Some(commit_object.clone()),
false,
@@ -3350,7 +3350,7 @@ impl SetDisks {
let mut fi = rename_commit.committed_file_info;
if needs_immediate_heal {
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
let mut request = rustfs_heal_contracts::heal_channel::create_heal_request_with_options(
commit_bucket.clone(),
Some(commit_object.clone()),
false,
@@ -3362,7 +3362,7 @@ impl SetDisks {
.or_else(|| commit_version_suspended.then(Uuid::nil))
.map(|version_id| version_id.to_string());
tokio::spawn(async move {
let _ = rustfs_common::heal_channel::send_heal_request(request).await;
let _ = rustfs_heal_contracts::heal_channel::send_heal_request(request).await;
});
}
@@ -7076,7 +7076,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
Some(scope),
);
}
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
let mut request = rustfs_heal_contracts::heal_channel::create_heal_request_with_options(
bucket.to_string(),
Some(object.to_string()),
false,
@@ -7085,7 +7085,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
Some(self.set_index),
);
request.object_version_id = (!version_id.is_empty()).then(|| version_id.to_string());
if let Err(e) = rustfs_common::heal_channel::send_heal_request(request).await {
if let Err(e) = rustfs_heal_contracts::heal_channel::send_heal_request(request).await {
warn!(
bucket,
object,
@@ -16546,7 +16546,7 @@ mod delete_objects_lock_gating_tests {
lifecycle_delete_all: Some(crate::object_api::LifecycleDeleteAllRequest {
version_id: Some(trigger_version_id),
delete_marker: false,
action: rustfs_common::metrics::IlmAction::DeleteAllVersionsAction,
action: rustfs_scanner_contracts::metrics::IlmAction::DeleteAllVersionsAction,
rule_id: "rule".to_string(),
phase: crate::object_api::LifecycleDeleteAllPhase::History,
}),
crates/ecstore/src/set_disk/read.rs
+10 -4
View File
@@ -1896,7 +1896,7 @@ fn is_get_object_metadata_cache_request_eligible(bucket: &str, opts: &ObjectOpti
#[cfg(test)]
mod metadata_cache_tests {
use super::*;
use rustfs_common::heal_channel::HealAdmissionDropReason;
use rustfs_heal_contracts::heal_channel::HealAdmissionDropReason;
use serial_test::serial;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Mutex, OnceLock};
@@ -1974,7 +1974,9 @@ mod metadata_cache_tests {
}
}
fn slow_read_repair_submitter(_request: rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture {
fn slow_read_repair_submitter(
_request: rustfs_heal_contracts::heal_channel::HealChannelRequest,
) -> ReadRepairAdmissionFuture {
SLOW_READ_REPAIR_SUBMITTER_CALLS.fetch_add(1, Ordering::Relaxed);
Box::pin(async {
tokio::time::sleep(Duration::from_millis(250)).await;
@@ -1982,14 +1984,18 @@ mod metadata_cache_tests {
})
}
fn dropped_read_repair_submitter(_request: rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture {
fn dropped_read_repair_submitter(
_request: rustfs_heal_contracts::heal_channel::HealChannelRequest,
) -> ReadRepairAdmissionFuture {
DROPPED_READ_REPAIR_SUBMITTER_CALLS.fetch_add(1, Ordering::Relaxed);
Box::pin(async {
ReadRepairAdmissionOutcome::Response(HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped))
})
}
fn capture_read_repair_submitter(request: rustfs_common::heal_channel::HealChannelRequest) -> ReadRepairAdmissionFuture {
fn capture_read_repair_submitter(
request: rustfs_heal_contracts::heal_channel::HealChannelRequest,
) -> ReadRepairAdmissionFuture {
CAPTURED_READ_REPAIR_CALLS.fetch_add(1, Ordering::Relaxed);
*CAPTURED_READ_REPAIR_PRIORITY.lock().expect("capture mutex poisoned") = Some(request.priority);
Box::pin(async {
crates/ecstore/src/store/heal.rs
+1 -1
View File
@@ -608,7 +608,7 @@ mod tests {
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations};
use crate::store::init_format::{load_format_erasure, save_format_file};
use crate::store::init_local_disks_with_instance_ctx;
use rustfs_common::heal_channel::DriveState;
use rustfs_heal_contracts::heal_channel::DriveState;
use tokio_util::sync::CancellationToken;
#[derive(Debug)]
crates/ecstore/src/store/init.rs
+2 -2
View File
@@ -9537,7 +9537,7 @@ mod tests {
.expect("unknown transition metadata should be written");
}
let lifecycle_event = crate::bucket::lifecycle::lifecycle::Event {
action: rustfs_common::metrics::IlmAction::DeleteAllVersionsAction,
action: rustfs_scanner_contracts::metrics::IlmAction::DeleteAllVersionsAction,
rule_id: "delete-all-versions".to_string(),
..Default::default()
};
@@ -9709,7 +9709,7 @@ mod tests {
lifecycle_delete_all: Some(crate::object_api::LifecycleDeleteAllRequest {
version_id: original.version_id,
delete_marker: false,
action: rustfs_common::metrics::IlmAction::DeleteAllVersionsAction,
action: rustfs_scanner_contracts::metrics::IlmAction::DeleteAllVersionsAction,
rule_id: "rule".to_string(),
phase: crate::object_api::LifecycleDeleteAllPhase::Preflight,
}),
crates/ecstore/src/store/mod.rs
+1 -1
View File
@@ -64,9 +64,9 @@ use futures::future::join_all;
use http::HeaderMap;
use lazy_static::lazy_static;
use rand::RngExt as _;
use rustfs_common::heal_channel::{HealItemType, HealOpts};
use rustfs_config::server_config::Config;
use rustfs_filemeta::FileInfo;
use rustfs_heal_contracts::heal_channel::{HealItemType, HealOpts};
use rustfs_lock::{LocalClient, LockClient, NamespaceLockWrapper};
use rustfs_madmin::heal_commands::HealResultItem;
use rustfs_utils::path::{decode_dir_object, encode_dir_object, path_join_buf};
crates/ecstore/src/store/rebalance.rs
+2 -2
View File
@@ -1049,7 +1049,7 @@ mod tests {
lifecycle_delete_all: Some(crate::object_api::LifecycleDeleteAllRequest {
version_id: Some(trigger_id),
delete_marker: false,
action: rustfs_common::metrics::IlmAction::DeleteAllVersionsAction,
action: rustfs_scanner_contracts::metrics::IlmAction::DeleteAllVersionsAction,
rule_id: "rule".to_string(),
phase: crate::object_api::LifecycleDeleteAllPhase::Preflight,
}),
@@ -1219,7 +1219,7 @@ mod tests {
lifecycle_delete_all: Some(crate::object_api::LifecycleDeleteAllRequest {
version_id: Some(marker_id),
delete_marker: true,
action: rustfs_common::metrics::IlmAction::DelMarkerDeleteAllVersionsAction,
action: rustfs_scanner_contracts::metrics::IlmAction::DelMarkerDeleteAllVersionsAction,
rule_id: "rule".to_string(),
phase: crate::object_api::LifecycleDeleteAllPhase::Preflight,
}),
crates/ecstore/tests/ecstore_contract_compat_test.rs
+1 -1
View File
@@ -14,8 +14,8 @@
mod storage_api;
use rustfs_common::heal_channel::HealOpts;
use rustfs_filemeta::FileInfo;
use rustfs_heal_contracts::heal_channel::HealOpts;
use rustfs_lock::NamespaceLockWrapper;
use rustfs_madmin::heal_commands::HealResultItem;
use storage_api::contract_compat::{