From bd5d27ffbdc5d5e959a192abf4848266561e10d8 Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 27 Jun 2025 17:32:30 +0800 Subject: [PATCH] fix --- crates/notify/src/integration.rs | 1 - crates/notify/src/lib.rs | 2 +- rustfs/src/main.rs | 28 +++++++++++++++------------- rustfs/src/server/service_state.rs | 6 +++--- rustfs/src/storage/ecfs.rs | 22 +++++++++++----------- 5 files changed, 30 insertions(+), 29 deletions(-) diff --git a/crates/notify/src/integration.rs b/crates/notify/src/integration.rs index 7fa9cfae3..ddd42240a 100644 --- a/crates/notify/src/integration.rs +++ b/crates/notify/src/integration.rs @@ -412,7 +412,6 @@ impl NotificationSystem { } } - // let rules_map = config.to_rules_map(); let rules_map = config.get_rules_map(); self.notifier.add_rules_map(bucket_name, rules_map.clone()).await; info!("Loaded notification config for bucket: {}", bucket_name); diff --git a/crates/notify/src/lib.rs b/crates/notify/src/lib.rs index 69183f0d9..450a3e2ad 100644 --- a/crates/notify/src/lib.rs +++ b/crates/notify/src/lib.rs @@ -26,7 +26,7 @@ pub use rules::BucketNotificationConfig; use std::io::IsTerminal; pub use target::Target; -use tracing_subscriber::{fmt, prelude::*, util::SubscriberInitExt, EnvFilter}; +use tracing_subscriber::{EnvFilter, fmt, prelude::*, util::SubscriberInitExt}; /// Initialize the tracing log system /// diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 5ff83cfac..b39582cca 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -12,10 +12,10 @@ mod service; mod storage; use crate::auth::IAMAuth; -use crate::console::{init_console_cfg, CONSOLE_CONFIG}; +use crate::console::{CONSOLE_CONFIG, init_console_cfg}; // Ensure the correct path for parse_license is imported use crate::event::shutdown_event_notifier; -use crate::server::{wait_for_shutdown, ServiceState, ServiceStateManager, ShutdownSignal, SHUTDOWN_TIMEOUT}; +use crate::server::{SHUTDOWN_TIMEOUT, ServiceState, ServiceStateManager, ShutdownSignal, wait_for_shutdown}; use bytes::Bytes; use chrono::Datelike; use clap::Parser; @@ -23,6 +23,7 @@ use common::{ // error::{Error, Result}, globals::set_global_addr, }; +use ecstore::StorageAPI; use ecstore::bucket::metadata_sys::init_bucket_metadata_sys; use ecstore::cmd::bucket_replication::init_bucket_replication_pool; use ecstore::config as ecconfig; @@ -30,12 +31,11 @@ use ecstore::config::GLOBAL_ConfigSys; use ecstore::heal::background_heal_ops::init_auto_heal; use ecstore::rpc::make_server; use ecstore::store_api::BucketOptions; -use ecstore::StorageAPI; use ecstore::{ endpoints::EndpointServerPools, heal::data_scanner::init_data_scanner, set_global_endpoints, - store::{init_local_disks, ECStore}, + store::{ECStore, init_local_disks}, update_erasure_type, }; use ecstore::{global::set_global_rustfs_port, notification_sys::new_global_notification_sys}; @@ -50,7 +50,7 @@ use iam::init_iam_sys; use license::init_license; use protos::proto_gen::node_service::node_service_server::NodeServiceServer; use rustfs_config::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY}; -use rustfs_obs::{init_obs, set_global_guard, SystemObserver}; +use rustfs_obs::{SystemObserver, init_obs, set_global_guard}; use rustfs_utils::net::parse_and_resolve_address; use rustls::ServerConfig; use s3s::{host::MultiDomain, service::S3ServiceBuilder}; @@ -62,12 +62,12 @@ use std::sync::Arc; use std::time::Duration; use tokio::net::TcpListener; #[cfg(unix)] -use tokio::signal::unix::{signal, SignalKind}; +use tokio::signal::unix::{SignalKind, signal}; use tokio_rustls::TlsAcceptor; -use tonic::{metadata::MetadataValue, Request, Status}; +use tonic::{Request, Status, metadata::MetadataValue}; use tower_http::cors::CorsLayer; use tower_http::trace::TraceLayer; -use tracing::{debug, error, info, instrument, warn, Span}; +use tracing::{Span, debug, error, info, instrument, warn}; const MI_B: usize = 1024 * 1024; @@ -610,11 +610,10 @@ async fn run(opt: config::Opt) -> Result<()> { tokio::time::sleep(SHUTDOWN_TIMEOUT).await; // listen to the shutdown signal match wait_for_shutdown().await { - ShutdownSignal::CtrlC - #[cfg(unix)] - | ShutdownSignal::Sigint - #[cfg(unix)] - | ShutdownSignal::Sigterm => { + signal + if matches!(signal, ShutdownSignal::CtrlC) + || (cfg!(unix) && matches!(signal, ShutdownSignal::Sigint | ShutdownSignal::Sigterm)) => + { info!("Shutdown signal received in main thread"); // update the status to stopping first state_manager.update(ServiceState::Stopping); @@ -630,6 +629,9 @@ async fn run(opt: config::Opt) -> Result<()> { state_manager.update(ServiceState::Stopped); info!("Server stopped current "); } + // This branch is required to make the match exhaustive. + // Because the guard above handles all variants, this branch should never be reached. + _ => unreachable!("All shutdown signals should be handled by the guarded arm"), } info!("server is stopped state: {:?}", state_manager.current_state()); diff --git a/rustfs/src/server/service_state.rs b/rustfs/src/server/service_state.rs index 4f936a294..a19afb822 100644 --- a/rustfs/src/server/service_state.rs +++ b/rustfs/src/server/service_state.rs @@ -1,6 +1,6 @@ use atomic_enum::atomic_enum; -use std::sync::atomic::Ordering; use std::sync::Arc; +use std::sync::atomic::Ordering; use std::time::Duration; use tracing::info; @@ -9,7 +9,7 @@ pub(crate) const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(1); #[cfg(target_os = "linux")] fn notify_systemd(state: &str) { - use libsystemd::daemon::{notify, NotifyState}; + use libsystemd::daemon::{NotifyState, notify}; use tracing::{debug, error}; let notify_state = match state { "ready" => NotifyState::Ready, @@ -53,7 +53,7 @@ pub(crate) enum ServiceState { #[cfg(unix)] pub(crate) async fn wait_for_shutdown() -> ShutdownSignal { - use tokio::signal::unix::{signal, SignalKind}; + use tokio::signal::unix::{SignalKind, signal}; let mut sigterm = signal(SignalKind::terminate()).expect("failed to create SIGTERM signal handler"); let mut sigint = signal(SignalKind::interrupt()).expect("failed to create SIGINT signal handler"); diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 36e70a46e..6e36ed95e 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -16,8 +16,8 @@ use bytes::Bytes; use chrono::DateTime; use chrono::Utc; use datafusion::arrow::csv::WriterBuilder as CsvWriterBuilder; -use datafusion::arrow::json::writer::JsonArray; use datafusion::arrow::json::WriterBuilder as JsonWriterBuilder; +use datafusion::arrow::json::writer::JsonArray; // use ecstore::store_api::RESERVED_METADATA_PREFIX; use ecstore::bucket::lifecycle::bucket_lifecycle_ops::validate_transition_tier; @@ -35,13 +35,13 @@ use ecstore::bucket::tagging::decode_tags; use ecstore::bucket::tagging::encode_tags; use ecstore::bucket::utils::serialize; use ecstore::bucket::versioning_sys::BucketVersioningSys; +use ecstore::cmd::bucket_replication::ReplicationStatusType; +use ecstore::cmd::bucket_replication::ReplicationType; use ecstore::cmd::bucket_replication::get_must_replicate_options; use ecstore::cmd::bucket_replication::must_replicate; use ecstore::cmd::bucket_replication::schedule_replication; -use ecstore::cmd::bucket_replication::ReplicationStatusType; -use ecstore::cmd::bucket_replication::ReplicationType; -use ecstore::compress::is_compressible; use ecstore::compress::MIN_COMPRESSIBLE_SIZE; +use ecstore::compress::is_compressible; use ecstore::error::StorageError; use ecstore::new_object_layer_fn; use ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; @@ -60,11 +60,11 @@ use futures::StreamExt; use http::HeaderMap; use lazy_static::lazy_static; use policy::auth; -use policy::policy::action::Action; -use policy::policy::action::S3Action; use policy::policy::BucketPolicy; use policy::policy::BucketPolicyArgs; use policy::policy::Validator; +use policy::policy::action::Action; +use policy::policy::action::S3Action; use query::instance::make_rustfsms; use rustfs_filemeta::headers::RESERVED_METADATA_PREFIX_LOWER; use rustfs_filemeta::headers::{AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING}; @@ -73,23 +73,23 @@ use rustfs_rio::EtagReader; use rustfs_rio::HashReader; use rustfs_rio::Reader; use rustfs_rio::WarpReader; -use rustfs_utils::path::path_join_buf; use rustfs_utils::CompressionAlgorithm; +use rustfs_utils::path::path_join_buf; use rustfs_zip::CompressionFormat; -use s3s::dto::*; -use s3s::s3_error; +use s3s::S3; use s3s::S3Error; use s3s::S3ErrorCode; use s3s::S3Result; -use s3s::S3; +use s3s::dto::*; +use s3s::s3_error; use s3s::{S3Request, S3Response}; use std::collections::HashMap; use std::fmt::Debug; use std::path::Path; use std::str::FromStr; use std::sync::Arc; -use time::format_description::well_known::Rfc3339; use time::OffsetDateTime; +use time::format_description::well_known::Rfc3339; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tokio_tar::Archive;