diff --git a/Cargo.lock b/Cargo.lock index 6579c4dcb..b0d00484b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3202,13 +3202,10 @@ dependencies = [ "base64 0.22.1", "bytes", "chrono", - "clap", "flatbuffers", "flate2", "futures", "http 1.4.0", - "http-body 1.0.1", - "http-body-util", "md5", "rand 0.10.1", "rcgen", diff --git a/crates/common/src/data_usage.rs b/crates/common/src/data_usage.rs index bb523ed31..21a8f9932 100644 --- a/crates/common/src/data_usage.rs +++ b/crates/common/src/data_usage.rs @@ -673,7 +673,7 @@ impl DataUsageCache { let mut leaves = Vec::new(); let mut remove = total - limit; add(self, path, &mut leaves); - leaves.sort_by(|a, b| a.objects.cmp(&b.objects)); + leaves.sort_by_key(|a| a.objects); while remove > 0 && !leaves.is_empty() { let Some(e) = leaves.first() else { diff --git a/crates/common/src/internode_metrics.rs b/crates/common/src/internode_metrics.rs index 2cae60253..025795817 100644 --- a/crates/common/src/internode_metrics.rs +++ b/crates/common/src/internode_metrics.rs @@ -101,11 +101,7 @@ impl InternodeMetrics { pub fn snapshot(&self) -> InternodeMetricsSnapshot { let dial_samples_total = self.dial_samples_total.load(Ordering::Relaxed); let dial_total_time_nanos = self.dial_total_time_nanos.load(Ordering::Relaxed); - let dial_avg_time_nanos = if dial_samples_total == 0 { - 0 - } else { - dial_total_time_nanos / dial_samples_total - }; + let dial_avg_time_nanos = dial_total_time_nanos.checked_div(dial_samples_total).unwrap_or(0); InternodeMetricsSnapshot { sent_bytes_total: self.sent_bytes_total.load(Ordering::Relaxed), diff --git a/crates/e2e_test/Cargo.toml b/crates/e2e_test/Cargo.toml index 8ade58c0f..ae5118148 100644 --- a/crates/e2e_test/Cargo.toml +++ b/crates/e2e_test/Cargo.toml @@ -41,7 +41,6 @@ serde_json.workspace = true tonic = { workspace = true } tokio = { workspace = true } tokio-stream = { workspace = true } -clap = { workspace = true } rustfs-madmin.workspace = true rustfs-filemeta.workspace = true bytes.workspace = true @@ -53,8 +52,6 @@ async-compression = { workspace = true, features = ["tokio", "bzip2", "xz"] } async-trait = { workspace = true } flate2.workspace = true http.workspace = true -http-body.workspace = true -http-body-util.workspace = true reqwest = { workspace = true } rustfs-signer.workspace = true tracing = { workspace = true } diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 72724ff02..c734eab17 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -1687,10 +1687,8 @@ pub async fn eval_action_from_lifecycle( let lock_enabled = if let Some(lr) = lr { lr.mode.is_some() } else { false }; match event.action { - IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => { - if lock_enabled { - return lifecycle::Event::default(); - } + IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction if lock_enabled => { + return lifecycle::Event::default(); } IlmAction::DeleteVersionAction | IlmAction::DeleteRestoredVersionAction => { if oi.version_id.is_none() { diff --git a/crates/ecstore/src/bucket/replication/replication_state.rs b/crates/ecstore/src/bucket/replication/replication_state.rs index 2efa3adba..dc2011df7 100644 --- a/crates/ecstore/src/bucket/replication/replication_state.rs +++ b/crates/ecstore/src/bucket/replication/replication_state.rs @@ -832,61 +832,55 @@ impl ReplicationStats { let mut rs = ReplStat::new(); match status { - ReplicationStatusType::Pending => { - if ri.op_type.is_data_replication() && prev_status != status { - rs.set( - ri.arn.clone(), - ri.size, - Duration::default(), - status, - ri.op_type, - ri.endpoint.clone(), - ri.secure, - ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())), - ); - } + ReplicationStatusType::Pending if ri.op_type.is_data_replication() && prev_status != status => { + rs.set( + ri.arn.clone(), + ri.size, + Duration::default(), + status, + ri.op_type, + ri.endpoint.clone(), + ri.secure, + ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())), + ); } - ReplicationStatusType::Completed => { - if ri.op_type.is_data_replication() { - rs.set( - ri.arn.clone(), - ri.size, - ri.duration, - status, - ri.op_type, - ri.endpoint.clone(), - ri.secure, - ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())), - ); - } + ReplicationStatusType::Completed if ri.op_type.is_data_replication() => { + rs.set( + ri.arn.clone(), + ri.size, + ri.duration, + status, + ri.op_type, + ri.endpoint.clone(), + ri.secure, + ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())), + ); } - ReplicationStatusType::Failed => { - if ri.op_type.is_data_replication() && prev_status == ReplicationStatusType::Pending { - rs.set( - ri.arn.clone(), - ri.size, - ri.duration, - status, - ri.op_type, - ri.endpoint.clone(), - ri.secure, - ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())), - ); - } + ReplicationStatusType::Failed + if ri.op_type.is_data_replication() && prev_status == ReplicationStatusType::Pending => + { + rs.set( + ri.arn.clone(), + ri.size, + ri.duration, + status, + ri.op_type, + ri.endpoint.clone(), + ri.secure, + ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())), + ); } - ReplicationStatusType::Replica => { - if ri.op_type == ReplicationType::Object { - rs.set( - ri.arn.clone(), - ri.size, - Duration::default(), - status, - ri.op_type, - String::new(), - false, - ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())), - ); - } + ReplicationStatusType::Replica if ri.op_type == ReplicationType::Object => { + rs.set( + ri.arn.clone(), + ri.size, + Duration::default(), + status, + ri.op_type, + String::new(), + false, + ri.error.as_ref().map(|e| crate::error::Error::other(e.clone())), + ); } _ => {} } diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 2f506a81c..c3e90d781 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -537,7 +537,7 @@ fn build_oidc_object(cfg: &Config) -> Map { }; let mut providers = subsystem.iter().collect::>(); - providers.sort_by(|(lhs, _), (rhs, _)| lhs.cmp(rhs)); + providers.sort_by_key(|(lhs, _)| *lhs); let mut oidc_obj = Map::new(); for (instance_key, kvs) in providers { @@ -572,7 +572,7 @@ fn build_semantic_oidc_object(cfg: &Config) -> Map { }; let mut providers = subsystem.iter().collect::>(); - providers.sort_by(|(lhs, _), (rhs, _)| lhs.cmp(rhs)); + providers.sort_by_key(|(lhs, _)| *lhs); let mut oidc_obj = Map::new(); for (instance_key, kvs) in providers { @@ -691,7 +691,7 @@ fn build_notify_subsystem_object( .iter() .filter(|(instance_key, _)| instance_key.as_str() != DEFAULT_DELIMITER) .collect::>(); - instances.sort_by(|(lhs, _), (rhs, _)| lhs.cmp(rhs)); + instances.sort_by_key(|(lhs, _)| *lhs); for (instance_key, kvs) in instances { let instance_obj = build_notify_instance_diff_object(kvs, &effective_default, valid_keys, default_kvs); diff --git a/crates/ecstore/src/pools.rs b/crates/ecstore/src/pools.rs index 8982f660b..02bd94632 100644 --- a/crates/ecstore/src/pools.rs +++ b/crates/ecstore/src/pools.rs @@ -1405,7 +1405,8 @@ impl ECStore { let mut fivs = load_decommission_entry_versions(&entry, &bucket, "file_info_versions")?; - fivs.versions.sort_by(|a, b| b.mod_time.cmp(&a.mod_time)); + fivs.versions + .sort_by_key(|v| (v.mod_time.is_none(), std::cmp::Reverse(v.mod_time))); let mut decommissioned: usize = 0; let mut expired: usize = 0; diff --git a/crates/ecstore/src/rebalance.rs b/crates/ecstore/src/rebalance.rs index 617142740..567958650 100644 --- a/crates/ecstore/src/rebalance.rs +++ b/crates/ecstore/src/rebalance.rs @@ -1530,7 +1530,8 @@ impl ECStore { let mut fivs = resolve_rebalance_file_info_versions_result(entry.file_info_versions(&bucket), bucket.as_str(), entry.name.as_str())?; - fivs.versions.sort_by(|a, b| b.mod_time.cmp(&a.mod_time)); + fivs.versions + .sort_by_key(|v| (v.mod_time.is_none(), std::cmp::Reverse(v.mod_time))); let mut rebalanced: usize = 0; let mut expired: usize = 0; diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 1c84edce4..dcc97a609 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -1627,7 +1627,7 @@ impl ObjectOperations for SetDisks { let mut vers = Vec::with_capacity(vers_map.len()); for (_, mut fi_vers) in vers_map { - fi_vers.versions.sort_by(|a, b| a.deleted.cmp(&b.deleted)); + fi_vers.versions.sort_by_key(|a| a.deleted); if let Some(index) = fi_vers.versions.iter().position(|fi| fi.deleted) { fi_vers.versions.truncate(index + 1); @@ -2961,7 +2961,7 @@ impl MultipartOperations for SetDisks { populated_upload_ids.insert(upload_id); } - uploads.sort_by(|a, b| a.initiated.cmp(&b.initiated)); + uploads.sort_by_key(|a| a.initiated); let mut upload_idx = 0; if let Some(upload_id_marker) = &upload_id_marker { @@ -4212,7 +4212,7 @@ async fn get_storage_info(disks: &[Option], eps: &[Endpoint]) -> rust // }, // } let mut disks = get_disks_info(disks, eps).await; - disks.sort_by(|a, b| a.total_space.cmp(&b.total_space)); + disks.sort_by_key(|a| a.total_space); // Provide minimal backend shape for callers. Do NOT guess parity here since it belongs to higher-level config. // Missing/empty standard_sc_data will be handled by capacity fallback logic. diff --git a/crates/ecstore/src/set_disk/lock.rs b/crates/ecstore/src/set_disk/lock.rs index b4711d2e5..4c8fad46d 100644 --- a/crates/ecstore/src/set_disk/lock.rs +++ b/crates/ecstore/src/set_disk/lock.rs @@ -168,7 +168,7 @@ impl SetDisks { let mut new_disks = Vec::new(); let mut new_infos = Vec::new(); - for (disk, info) in disks.into_iter().zip(infos.into_iter()) { + for (disk, info) in disks.into_iter().zip(infos) { let Some(info) = info else { continue; }; diff --git a/crates/ecstore/src/store_list_objects.rs b/crates/ecstore/src/store_list_objects.rs index 915670dc7..98a031c36 100644 --- a/crates/ecstore/src/store_list_objects.rs +++ b/crates/ecstore/src/store_list_objects.rs @@ -162,10 +162,8 @@ impl ListPathOptions { } match kv[0] { - "rustfs_cache" => { - if kv[1] != MARKER_TAG_VERSION { - continue; - } + "rustfs_cache" if kv[1] != MARKER_TAG_VERSION => { + continue; } "id" => self.id = Some(kv[1].to_owned()), "return" => { diff --git a/crates/filemeta/examples/dump_versions.rs b/crates/filemeta/examples/dump_versions.rs index f9dd8e39c..bfceac35a 100644 --- a/crates/filemeta/examples/dump_versions.rs +++ b/crates/filemeta/examples/dump_versions.rs @@ -1,3 +1,17 @@ +// 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 rustfs_filemeta::FileMeta; use std::{env, fs, path::PathBuf}; diff --git a/crates/filemeta/src/fileinfo.rs b/crates/filemeta/src/fileinfo.rs index d5061fe0c..353aba8c1 100644 --- a/crates/filemeta/src/fileinfo.rs +++ b/crates/filemeta/src/fileinfo.rs @@ -349,7 +349,7 @@ impl FileInfo { self.parts.push(part); - self.parts.sort_by(|a, b| a.number.cmp(&b.number)); + self.parts.sort_by_key(|a| a.number); } // to_part_offset gets the part index where offset is located, returns part index and offset @@ -643,13 +643,11 @@ pub fn parse_restore_obj_status(restore_hdr: &str) -> Result { } match progress_tokens[1] { - "true" | "\"true\"" => { - if tokens.len() == 1 { - return Ok(RestoreStatus { - is_restore_in_progress: Some(true), - ..Default::default() - }); - } + "true" | "\"true\"" if tokens.len() == 1 => { + return Ok(RestoreStatus { + is_restore_in_progress: Some(true), + ..Default::default() + }); } "false" | "\"false\"" => { if tokens.len() != 2 { diff --git a/crates/filemeta/src/test_data.rs b/crates/filemeta/src/test_data.rs index 704e9d2f8..157595b08 100644 --- a/crates/filemeta/src/test_data.rs +++ b/crates/filemeta/src/test_data.rs @@ -104,7 +104,8 @@ pub fn create_real_xlmeta() -> Result> { fm.versions.push(legacy_shallow); // Sort by modification time (newest first) - fm.versions.sort_by(|a, b| b.header.mod_time.cmp(&a.header.mod_time)); + fm.versions + .sort_by_key(|v| (v.header.mod_time.is_none(), std::cmp::Reverse(v.header.mod_time))); fm.marshal_msg() } @@ -263,11 +264,11 @@ pub fn create_legacy_v1_object_xlmeta() -> Result> { wr.extend_from_slice(&[0xc6, 0, 0, 0, 0]); let offset = wr.len(); - rmp::encode::write_uint(&mut wr, 1).unwrap(); - rmp::encode::write_uint(&mut wr, 1).unwrap(); - rmp::encode::write_sint(&mut wr, 1).unwrap(); - rmp::encode::write_bin(&mut wr, &header).unwrap(); - rmp::encode::write_bin(&mut wr, &body).unwrap(); + rmp::encode::write_uint(&mut wr, 1)?; + rmp::encode::write_uint(&mut wr, 1)?; + rmp::encode::write_sint(&mut wr, 1)?; + rmp::encode::write_bin(&mut wr, &header)?; + rmp::encode::write_bin(&mut wr, &body)?; let data_len = (wr.len() - offset) as u32; wr[offset - 4..offset].copy_from_slice(&data_len.to_be_bytes()); @@ -350,7 +351,8 @@ pub fn create_complex_xlmeta() -> Result> { } // Sort by modification time (newest first) - fm.versions.sort_by(|a, b| b.header.mod_time.cmp(&a.header.mod_time)); + fm.versions + .sort_by_key(|v| (v.header.mod_time.is_none(), std::cmp::Reverse(v.header.mod_time))); fm.marshal_msg() } diff --git a/crates/io-core/src/lock_optimizer.rs b/crates/io-core/src/lock_optimizer.rs index 08ae2d8fe..3ad73f3ca 100644 --- a/crates/io-core/src/lock_optimizer.rs +++ b/crates/io-core/src/lock_optimizer.rs @@ -124,11 +124,7 @@ impl LockStats { pub fn avg_hold_time(&self) -> Duration { let total = self.total_hold_time_ns.load(Ordering::Relaxed); let count = self.locks_acquired.load(Ordering::Relaxed); - if count == 0 { - Duration::ZERO - } else { - Duration::from_nanos(total / count) - } + total.checked_div(count).map(Duration::from_nanos).unwrap_or(Duration::ZERO) } /// Get maximum hold time. diff --git a/crates/io-core/src/timeout_wrapper.rs b/crates/io-core/src/timeout_wrapper.rs index 6af86f789..426911925 100644 --- a/crates/io-core/src/timeout_wrapper.rs +++ b/crates/io-core/src/timeout_wrapper.rs @@ -342,11 +342,7 @@ impl TimeoutStats { pub fn avg_wait_time(&self) -> Duration { let total = self.total_wait_time_ns.load(Ordering::Relaxed); let count = self.total_operations.load(Ordering::Relaxed); - if count == 0 { - Duration::ZERO - } else { - Duration::from_nanos(total / count) - } + total.checked_div(count).map(Duration::from_nanos).unwrap_or(Duration::ZERO) } /// Reset statistics. diff --git a/crates/io-metrics/src/adaptive_ttl.rs b/crates/io-metrics/src/adaptive_ttl.rs index ceac9f853..50ddc7ebd 100644 --- a/crates/io-metrics/src/adaptive_ttl.rs +++ b/crates/io-metrics/src/adaptive_ttl.rs @@ -255,7 +255,7 @@ impl AccessTracker { /// Get keys sorted by access count (descending). pub fn top_keys(&self, n: usize) -> Vec<(&String, &AccessRecord)> { let mut entries: Vec<_> = self.records.iter().collect(); - entries.sort_by(|a, b| b.1.count.cmp(&a.1.count)); + entries.sort_by_key(|entry| std::cmp::Reverse(entry.1.count)); entries.into_iter().take(n).collect() } diff --git a/crates/lock/src/fast_lock/metrics.rs b/crates/lock/src/fast_lock/metrics.rs index 03a5940f8..cc6a19b29 100644 --- a/crates/lock/src/fast_lock/metrics.rs +++ b/crates/lock/src/fast_lock/metrics.rs @@ -170,11 +170,10 @@ impl MetricsSnapshot { } pub fn avg_wait_time(&self) -> Duration { - if self.slow_path_success == 0 { - Duration::ZERO - } else { - Duration::from_nanos(self.total_wait_time_ns / self.slow_path_success) - } + self.total_wait_time_ns + .checked_div(self.slow_path_success) + .map(Duration::from_nanos) + .unwrap_or(Duration::ZERO) } pub fn max_wait_time(&self) -> Duration { diff --git a/crates/object-capacity/src/scan.rs b/crates/object-capacity/src/scan.rs index 9fab908ab..dae73cad8 100644 --- a/crates/object-capacity/src/scan.rs +++ b/crates/object-capacity/src/scan.rs @@ -603,11 +603,7 @@ async fn get_dir_size_async(path: &Path) -> Result max_files_threshold { let overflow_count = file_count - max_files_threshold; let exact_prefix_count = file_count.min(max_files_threshold) as u64; - let avg_prefix_size = if exact_prefix_count > 0 { - exact_prefix_bytes / exact_prefix_count - } else { - 0 - }; + let avg_prefix_size = exact_prefix_bytes.checked_div(exact_prefix_count).unwrap_or(0); let estimated_overflow = avg_prefix_size.saturating_mul(overflow_count as u64); let estimated_size = exact_prefix_bytes.saturating_add(estimated_overflow); info!( diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 11b01016c..c5b2ca952 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -765,7 +765,7 @@ impl DataUsageCache { let mut leaves = Vec::new(); let mut remove = total - limit; add(self, path, &mut leaves); - leaves.sort_by(|a, b| a.objects.cmp(&b.objects)); + leaves.sort_by_key(|a| a.objects); while remove > 0 && !leaves.is_empty() { let e = leaves.first().unwrap(); diff --git a/rustfs/src/admin/handlers/policies.rs b/rustfs/src/admin/handlers/policies.rs index dd471daf8..45d0e5862 100644 --- a/rustfs/src/admin/handlers/policies.rs +++ b/rustfs/src/admin/handlers/policies.rs @@ -641,8 +641,8 @@ fn build_policy_mappings( } let mut results: Vec = policy_map - .into_iter() - .filter_map(|(_, mut mapping)| { + .into_values() + .filter_map(|mut mapping| { if !requested_policies.is_empty() && !requested_policies.iter().any(|policy| policy == &mapping.policy) { return None; } diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs index 218cdbac2..247abb296 100644 --- a/rustfs/src/embedded.rs +++ b/rustfs/src/embedded.rs @@ -28,8 +28,8 @@ //! let port = find_available_port()?; //! let server = RustFSServerBuilder::new() //! .address(format!("127.0.0.1:{port}")) -//! .access_key("minioadmin") -//! .secret_key("minioadmin") +//! .access_key("rustfsadmin") +//! .secret_key("rustfsadmin") //! .build() //! .await?; //! @@ -49,10 +49,7 @@ use crate::app::context::{AppContext, init_global_app_context}; use crate::config::Config; use crate::init::{add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system}; -use crate::server::{ - ServiceState, ServiceStateManager, init_event_notifier, shutdown_event_notifier, start_audit_system, start_http_server, - stop_audit_system, -}; +use crate::server::{init_event_notifier, shutdown_event_notifier, start_audit_system, start_http_server, stop_audit_system}; use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr}; use rustfs_credentials::init_global_action_credentials; use rustfs_ecstore::store::init_lock_clients; @@ -166,7 +163,7 @@ impl RustFSServerBuilder { /// /// Defaults: /// - address: `"127.0.0.1:9000"` - /// - access_key / secret_key: `"minioadmin"` + /// - access_key / secret_key: `"rustfsadmin"` /// - region: `"us-east-1"` /// - A temporary directory is created automatically for data storage /// @@ -175,10 +172,10 @@ impl RustFSServerBuilder { pub fn new() -> Self { Self { address: "127.0.0.1:9000".to_string(), - access_key: "minioadmin".to_string(), - secret_key: "minioadmin".to_string(), + access_key: rustfs_credentials::DEFAULT_ACCESS_KEY.to_string(), + secret_key: rustfs_credentials::DEFAULT_SECRET_KEY.to_string(), volumes: Vec::new(), - region: "us-east-1".to_string(), + region: rustfs_config::RUSTFS_REGION.to_string(), } } @@ -196,13 +193,13 @@ impl RustFSServerBuilder { self } - /// Set the S3 access key (default: `"minioadmin"`). + /// Set the S3 access key (default: `"rustfsadmin"`). pub fn access_key(mut self, key: impl Into) -> Self { self.access_key = key.into(); self } - /// Set the S3 secret key (default: `"minioadmin"`). + /// Set the S3 secret key (default: `"rustfsadmin"`). pub fn secret_key(mut self, key: impl Into) -> Self { self.secret_key = key.into(); self @@ -355,13 +352,11 @@ impl RustFSServerBuilder { // Service state. let readiness = Arc::new(GlobalReadiness::new()); - let state_manager = ServiceStateManager::new(); - state_manager.update(ServiceState::Starting); // Start HTTP server. let mut s3_config = config.clone(); s3_config.console_enable = false; - let (shutdown_tx, bound_addr) = start_http_server(&s3_config, state_manager.clone(), readiness.clone()).await?; + let (shutdown_tx, bound_addr) = start_http_server(&s3_config, readiness.clone()).await?; let ctx = CancellationToken::new(); let shutdown_embedded_server = || { let _ = shutdown_tx.send(()); @@ -599,15 +594,15 @@ impl Drop for RustFSServer { /// ```rust,no_run /// use rustfs::embedded::{find_available_port, RustFSServerBuilder}; /// -/// # async fn example() -> Result<(), Box> { -/// let port = find_available_port()?; -/// let server = RustFSServerBuilder::new() -/// .address(format!("127.0.0.1:{port}")) -/// .build() -/// .await?; -/// println!("Listening on port {port}"); -/// # Ok(()) -/// # } +/// async fn example() -> Result<(), Box> { +/// let port = find_available_port()?; +/// let server = RustFSServerBuilder::new() +/// .address(format!("127.0.0.1:{port}")) +/// .build() +/// .await?; +/// println!("Listening on port {port}"); +/// Ok(()) +/// } /// ``` pub fn find_available_port() -> Result { let listener = std::net::TcpListener::bind("127.0.0.1:0")?; diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 9a5b2dee7..cdb9eed8b 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -340,14 +340,14 @@ async fn run(config: rustfs::config::Config) -> Result<()> { let s3_shutdown_tx = { let mut s3_config = config.clone(); s3_config.console_enable = false; - let (s3_shutdown_tx, _) = start_http_server(&s3_config, state_manager.clone(), readiness.clone()).await?; + let (s3_shutdown_tx, _) = start_http_server(&s3_config, readiness.clone()).await?; Some(s3_shutdown_tx) }; let console_shutdown_tx = if config.console_enable && !config.console_address.is_empty() { let mut console_config = config.clone(); console_config.address = console_config.console_address.clone(); - let (console_shutdown_tx, _) = start_http_server(&console_config, state_manager.clone(), readiness.clone()).await?; + let (console_shutdown_tx, _) = start_http_server(&console_config, readiness.clone()).await?; Some(console_shutdown_tx) } else { None @@ -576,11 +576,11 @@ async fn run(config: rustfs::config::Config) -> Result<()> { ); // 4. Mark as Full Ready now that critical components are warm readiness.mark_stage(SystemStage::FullReady); - // Update service status to Ready - state_manager.update(ServiceState::Ready); // Set the global RustFS initialization time to now rustfs_common::set_global_init_time_now().await; + // Publish ready only after all critical bootstrap metadata is in place + state_manager.update(ServiceState::Ready); // Perform hibernation for 1 second tokio::time::sleep(SHUTDOWN_TIMEOUT).await; diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 23461d30e..3fad55f40 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -18,7 +18,7 @@ use crate::auth::IAMAuth; use crate::auth_keystone; use crate::config; use crate::server::{ - ReadinessGateLayer, RemoteAddr, ServiceState, ServiceStateManager, + ReadinessGateLayer, RemoteAddr, compress::{CompressionConfig, PathAwareCompressionPredicate, PathCategoryInjectionLayer}, hybrid::hybrid, layer::{ @@ -68,7 +68,6 @@ use tracing_opentelemetry::OpenTelemetrySpanExt; pub async fn start_http_server( config: &config::Config, - worker_state_manager: ServiceStateManager, readiness: Arc, ) -> Result<(tokio::sync::broadcast::Sender<()>, SocketAddr)> { let server_addr = parse_and_resolve_address(config.address.as_str()).map_err(Error::other)?; @@ -161,12 +160,12 @@ pub async fn start_http_server( // Note: outbound material (root CAs, mTLS identity) is already applied in main.rs. let tls_snapshot = TlsMaterialSnapshot::load(tls_path) .await - .map_err(|e| std::io::Error::other(e.to_string()))?; + .map_err(|e| Error::other(e.to_string()))?; let tls_acceptor = tls_snapshot .build_tls_acceptor(tls_path) .await - .map_err(|e| std::io::Error::other(e.to_string()))?; + .map_err(|e| Error::other(e.to_string()))?; let tls_enabled = tls_acceptor.is_some(); let protocol = if tls_enabled { "https" } else { "http" }; @@ -357,10 +356,6 @@ pub async fn start_http_server( let graceful = Arc::new(GracefulShutdown::new()); debug!("graceful initiated"); - // service ready - worker_state_manager.update(ServiceState::Ready); - // tls_acceptor is already Option>, clone for the loop - loop { debug!("Waiting for new connection..."); let (socket, _) = { @@ -458,7 +453,6 @@ pub async fn start_http_server( process_connection(socket, tls_acceptor.clone(), connection_ctx, graceful.clone()); } - worker_state_manager.update(ServiceState::Stopping); match Arc::try_unwrap(graceful) { Ok(g) => { tokio::select! { @@ -476,7 +470,6 @@ pub async fn start_http_server( debug!("Timeout reached, forcing shutdown"); } } - worker_state_manager.update(ServiceState::Stopped); }); Ok((shutdown_tx, local_addr)) @@ -490,7 +483,7 @@ struct ConnectionContext { is_console: bool, readiness: Arc, /// Pre-computed Keystone auth provider (avoids per-connection OnceLock read). - keystone_auth: Option>, + keystone_auth: Option>, /// Pre-computed trusted proxy layer (avoids per-connection is_enabled() check). trusted_proxy_layer: Option, } diff --git a/rustfs/src/server/service_state.rs b/rustfs/src/server/service_state.rs index 420ab3264..be400c2ba 100644 --- a/rustfs/src/server/service_state.rs +++ b/rustfs/src/server/service_state.rs @@ -13,39 +13,18 @@ // limitations under the License. use atomic_enum::atomic_enum; -use std::sync::Arc; use std::sync::atomic::Ordering; +use std::sync::{Arc, Mutex}; use std::time::Duration; -use tracing::info; +use tracing::{info, warn}; // a configurable shutdown timeout pub const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(1); -#[cfg(target_os = "linux")] -fn notify_systemd(state: &str) { - use libsystemd::daemon::{NotifyState, notify}; - use tracing::{debug, error}; - let notify_state = match state { - "ready" => NotifyState::Ready, - "stopping" => NotifyState::Stopping, - _ => { - info!("Unsupported state passed to notify_systemd: {}", state); - return; - } - }; - - if let Err(e) = notify(false, &[notify_state]) { - error!("Failed to notify systemd: {}", e); - } else { - debug!("Successfully notified systemd: {}", state); - } - info!("Systemd notifications are enabled on linux (state: {})", state); -} - -#[cfg(not(target_os = "linux"))] -fn notify_systemd(state: &str) { - info!("Systemd notifications are not available on this platform not linux (state: {})", state); -} +const SERVICE_STATUS_STARTING: &str = "Starting"; +const SERVICE_STATUS_RUNNING: &str = "Running"; +const SERVICE_STATUS_STOPPING: &str = "Stopping"; +const SERVICE_STATUS_STOPPED: &str = "Stopped"; #[derive(Debug)] pub enum ShutdownSignal { @@ -100,56 +79,112 @@ pub async fn wait_for_shutdown() -> ShutdownSignal { #[derive(Clone)] pub struct ServiceStateManager { state: Arc, + published_state: Arc>>, } impl ServiceStateManager { pub fn new() -> Self { Self { state: Arc::new(AtomicServiceState::new(ServiceState::Starting)), + published_state: Arc::new(Mutex::new(None)), } } pub fn update(&self, new_state: ServiceState) { + // Serialize transition check + state write + publish dedupe + notify as one + // critical section to keep notification order monotonic under concurrency. + let mut published_state = self.published_state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let current_state = self.current_state(); + if service_state_rank(new_state) < service_state_rank(current_state) { + warn!( + current = ?current_state, + attempted = ?new_state, + "Ignoring regressive service state transition" + ); + return; + } + self.state.store(new_state, Ordering::SeqCst); - self.notify_systemd(&new_state); + + if *published_state != Some(new_state) { + *published_state = Some(new_state); + self.notify_systemd(new_state); + } } pub fn current_state(&self) -> ServiceState { self.state.load(Ordering::SeqCst) } - fn notify_systemd(&self, state: &ServiceState) { + fn notify_systemd(&self, state: ServiceState) { match state { ServiceState::Starting => { info!("RustFS Service is starting..."); - #[cfg(target_os = "linux")] - if let Err(e) = - libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Starting...".to_string())]) - { - tracing::error!("Failed to notify systemd of starting state: {}", e); - } + notify_systemd_daemon(state); } ServiceState::Ready => { - info!("RustFS Service is ready"); - notify_systemd("ready"); + info!("RustFS Service is running"); + notify_systemd_daemon(state); } ServiceState::Stopping => { info!("RustFS Service is stopping..."); - notify_systemd("stopping"); + notify_systemd_daemon(state); } ServiceState::Stopped => { info!("RustFS Service has stopped"); - #[cfg(target_os = "linux")] - if let Err(e) = - libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Stopped".to_string())]) - { - tracing::error!("Failed to notify systemd of stopped state: {}", e); - } + notify_systemd_daemon(state); } } } } +fn service_state_rank(state: ServiceState) -> u8 { + match state { + ServiceState::Starting => 0, + ServiceState::Ready => 1, + ServiceState::Stopping => 2, + ServiceState::Stopped => 3, + } +} + +fn systemd_status_text(state: ServiceState) -> &'static str { + match state { + ServiceState::Starting => SERVICE_STATUS_STARTING, + ServiceState::Ready => SERVICE_STATUS_RUNNING, + ServiceState::Stopping => SERVICE_STATUS_STOPPING, + ServiceState::Stopped => SERVICE_STATUS_STOPPED, + } +} + +#[cfg(target_os = "linux")] +fn notify_systemd_daemon(state: ServiceState) { + use libsystemd::daemon::{NotifyState, notify}; + use tracing::{debug, error}; + + let status = systemd_status_text(state); + let result = match state { + ServiceState::Starting => notify(false, &[NotifyState::Status(status.to_string())]), + ServiceState::Ready => notify(false, &[NotifyState::Ready, NotifyState::Status(status.to_string())]), + ServiceState::Stopping => notify(false, &[NotifyState::Stopping, NotifyState::Status(status.to_string())]), + ServiceState::Stopped => notify(false, &[NotifyState::Status(status.to_string())]), + }; + + if let Err(e) = result { + error!(%status, ?state, "Failed to notify systemd: {}", e); + } else { + debug!(%status, ?state, "Successfully notified systemd"); + } +} + +#[cfg(not(target_os = "linux"))] +fn notify_systemd_daemon(state: ServiceState) { + info!( + status = systemd_status_text(state), + ?state, + "Systemd notifications are not available on this platform" + ); +} + impl Default for ServiceStateManager { fn default() -> Self { Self::new() @@ -180,4 +215,23 @@ mod tests { manager.update(ServiceState::Stopped); assert_eq!(manager.current_state(), ServiceState::Stopped); } + + #[test] + fn test_service_state_manager_ignores_regression() { + let manager = ServiceStateManager::new(); + + manager.update(ServiceState::Starting); + manager.update(ServiceState::Ready); + manager.update(ServiceState::Starting); + assert_eq!(manager.current_state(), ServiceState::Ready); + + manager.update(ServiceState::Stopping); + manager.update(ServiceState::Ready); + assert_eq!(manager.current_state(), ServiceState::Stopping); + } + + #[test] + fn test_ready_maps_to_running_status() { + assert_eq!(systemd_status_text(ServiceState::Ready), SERVICE_STATUS_RUNNING); + } } diff --git a/rustfs/src/storage/concurrency/io_schedule.rs b/rustfs/src/storage/concurrency/io_schedule.rs index 53c74570b..59a2a689f 100644 --- a/rustfs/src/storage/concurrency/io_schedule.rs +++ b/rustfs/src/storage/concurrency/io_schedule.rs @@ -1263,11 +1263,7 @@ impl IoLoadMetrics { pub(crate) fn lifetime_average_wait(&self) -> Duration { let total = self.total_wait_ns.load(Ordering::Relaxed); let count = self.observation_count.load(Ordering::Relaxed); - if count == 0 { - Duration::ZERO - } else { - Duration::from_nanos(total / count) - } + total.checked_div(count).map(Duration::from_nanos).unwrap_or(Duration::ZERO) } /// Get the total observation count diff --git a/rustfs/src/storage/lock_optimizer.rs b/rustfs/src/storage/lock_optimizer.rs index 3b7859c29..5c6255ea6 100644 --- a/rustfs/src/storage/lock_optimizer.rs +++ b/rustfs/src/storage/lock_optimizer.rs @@ -147,11 +147,7 @@ impl LockStats { pub fn avg_hold_time(&self) -> Duration { let total = self.total_hold_time_us.load(Ordering::Relaxed); let count = self.locks_released_early.load(Ordering::Relaxed); - if count > 0 { - Duration::from_micros(total / count) - } else { - Duration::ZERO - } + total.checked_div(count).map(Duration::from_micros).unwrap_or(Duration::ZERO) } /// Get maximum hold time.