diff --git a/crates/event/src/adapter/kafka.rs b/crates/event/src/adapter/kafka.rs index a60afaeee..c480ef23f 100644 --- a/crates/event/src/adapter/kafka.rs +++ b/crates/event/src/adapter/kafka.rs @@ -1,7 +1,8 @@ -use crate::Event; +use crate::config::{default_queue_limit, STORE_PREFIX}; use crate::KafkaConfig; use crate::{ChannelAdapter, ChannelAdapterType}; use crate::{Error, QueueStore}; +use crate::{Event, DEFAULT_RETRY_INTERVAL}; use async_trait::async_trait; use rdkafka::error::KafkaError; use rdkafka::producer::{FutureProducer, FutureRecord}; @@ -11,6 +12,7 @@ use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; use tokio::time::sleep; +use ChannelAdapterType::Kafka; /// Kafka adapter for sending events to a Kafka topic. pub struct KafkaAdapter { @@ -26,12 +28,24 @@ impl KafkaAdapter { let producer = rdkafka::config::ClientConfig::new() .set("bootstrap.servers", &config.brokers) .set("message.timeout.ms", config.timeout.to_string()) - .create()?; + .create() + .map_err(|e| Error::msg(format!("Failed to create a Kafka producer: {}", e)))?; // create a queue store if enabled - let store = if !config.queue_dir.is_empty() { - let store_path = PathBuf::from(&config.queue_dir); - let store = QueueStore::new(store_path, config.queue_limit, Some(".kafka".to_string())); + let store = if !config.common.queue_dir.is_empty() { + let store_path = PathBuf::from(&config.common.queue_dir).join(format!( + "{}-{}-{}", + STORE_PREFIX, + Kafka.as_str(), + config.common.identifier + )); + + let queue_limit = if config.queue_limit > 0 { + config.queue_limit + } else { + default_queue_limit() + }; + let store = QueueStore::new(store_path, config.queue_limit, Some(".event".to_string())); if let Err(e) = store.open() { tracing::error!("Unable to open queue storage: {}", e); None @@ -44,19 +58,58 @@ impl KafkaAdapter { Ok(Self { config, producer, store }) } + + /// Handle backlog events in storage + pub async fn process_backlog(&self) -> Result<(), Error> { + if let Some(store) = &self.store { + let keys = store.list(); + + for key in keys { + match store.get_multiple(&key) { + Ok(events) => { + for event in events { + // Use the retry interval to send events + if let Err(e) = self.send_with_retry(&event).await { + tracing::error!("Processing of backlog events failed: {}", e); + // If it still fails, we remain in the queue + break; + } + } + + // The event is deleted after it has been successfully processed + if let Err(e) = store.del(&key) { + tracing::error!("Failed to delete a handled event: {}", e); + } + } + Err(e) => { + tracing::error!("Fetch events from the queue failed: {}", e); + + // If the event cannot be read, it may be corrupted, delete it + if let Err(del_err) = store.del(&key) { + tracing::error!("Failed to delete a corrupted event: {}", del_err); + } + } + } + } + } + + Ok(()) + } + /// Sends an event to the Kafka topic with retry logic. async fn send_with_retry(&self, event: &Event) -> Result<(), Error> { - let event_id = event.id.to_string(); - let payload = serde_json::to_string(&event)?; + let retry_interval = match self.config.retry_interval { + Some(t) => Duration::from_secs(t), + None => Duration::from_secs(DEFAULT_RETRY_INTERVAL), // Default to 3 seconds if not set + }; for attempt in 0..self.max_retries { - let record = FutureRecord::to(&self.topic).key(&event_id).payload(&payload); - - match self.producer.send(record, Timeout::Never).await { + match self.send_request(event).await { Ok(_) => return Ok(()), Err((KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull), _)) => { tracing::warn!("Kafka attempt {} failed: Queue full. Retrying...", attempt + 1); - sleep(Duration::from_secs(2u64.pow(attempt))).await; + // sleep(Duration::from_secs(2u64.pow(attempt))).await; + sleep(retry_interval).await; } Err((e, _)) => { tracing::error!("Kafka send error: {}", e); @@ -67,6 +120,38 @@ impl KafkaAdapter { Err(Error::Custom("Exceeded maximum retry attempts for Kafka message".to_string())) } + + /// Send a single Kafka message + async fn send_request(&self, event: &Event) -> Result<(), Error> { + // Serialize events + let payload = serde_json::to_string(event).map_err(|e| Error::Custom(format!("Serialization event failed: {}", e)))?; + + // Create a Kafka record + let record = FutureRecord::to(&self.config.topic).payload(&payload).key(&event.id); // Use the event ID as the key + + // Send to Kafka + let delivery_status = self + .producer + .send(record, Duration::from_millis(self.config.timeout)) + .await + .map_err(|(e, _)| Error::Custom(format!("Failed to send to Kafka: {}", e)))?; + // Check delivery status + if let Some((err, _)) = delivery_status { + return Err(Error::Kafka(err)); + } + + Ok(()) + } + + /// Save the event to the queue + async fn save_to_queue(&self, event: &Event) -> Result<(), Error> { + if let Some(store) = &self.store { + store + .put(event.clone()) + .map_err(|e| Error::Custom(format!("Saving events to queue failed: {}", e)))?; + } + Ok(()) + } } #[async_trait] @@ -76,6 +161,21 @@ impl ChannelAdapter for KafkaAdapter { } async fn send(&self, event: &Event) -> Result<(), Error> { - self.send_with_retry(event).await + // Try to deal with the backlog of events first + let _ = self.process_backlog().await; + + // An attempt was made to send the current event + match self.send_with_retry(event).await { + Ok(_) => Ok(()), + Err(e) => { + // If the send fails and the queue is enabled, save to the queue + if let Some(_) = &self.store { + tracing::warn!("Failed to send events to Kafka and saved to a queue: {}", e); + self.save_to_queue(event).await?; + return Ok(()); + } + Err(e) + } + } } } diff --git a/crates/event/src/adapter/webhook.rs b/crates/event/src/adapter/webhook.rs index 1de3e40d7..22884646d 100644 --- a/crates/event/src/adapter/webhook.rs +++ b/crates/event/src/adapter/webhook.rs @@ -1,20 +1,26 @@ +use crate::config::STORE_PREFIX; use crate::store::queue::Store; use crate::WebhookConfig; use crate::{ChannelAdapter, ChannelAdapterType}; use crate::{Error, QueueStore}; use crate::{Event, DEFAULT_RETRY_INTERVAL}; use async_trait::async_trait; +use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; use reqwest::{self, Client, Identity, RequestBuilder}; use std::fs; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; use tokio::time::sleep; +use ChannelAdapterType::Webhook; /// Webhook adapter for sending events to a webhook endpoint. pub struct WebhookAdapter { + /// Configuration information config: WebhookConfig, + /// Event storage queues store: Option>>, + /// HTTP client client: Client, } @@ -65,12 +71,25 @@ impl WebhookAdapter { } else { builder.build() } - .expect("Failed to create HTTP client"); + .unwrap_or_else(|e| { + tracing::error!("Failed to create HTTP client: {}", e); + reqwest::Client::new() + }); // create a queue store if enabled let store = if !config.common.queue_dir.len() > 0 { - let store_path = PathBuf::from(&config.common.queue_dir); - let store = QueueStore::new(store_path, config.common.queue_limit, Some(".webhook".to_string())); + let store_path = PathBuf::from(&config.common.queue_dir).join(format!( + "{}-{}-{}", + STORE_PREFIX, + Webhook.as_str(), + config.common.identifier + )); + let queue_limit = if config.common.queue_limit > 0 { + config.common.queue_limit + } else { + crate::config::default_queue_limit() + }; + let store = QueueStore::new(store_path, queue_limit, Some(".event".to_string())); if let Err(e) = store.open() { tracing::error!("Unable to open queue storage: {}", e); None @@ -94,16 +113,22 @@ impl WebhookAdapter { for event in events { if let Err(e) = self.send_with_retry(&event).await { tracing::error!("Processing of backlog events failed: {}", e); - continue; + // If it still fails, we remain in the queue + break; } } // Deleted after successful processing - let _ = store.del(&key); + if let Err(e) = store.del(&key) { + tracing::error!("Failed to delete a handled event: {}", e); + } } Err(e) => { tracing::error!("Failed to read events from storage: {}", e); // delete the broken entries - let _ = store.del(&key); + // If the event cannot be read, it may be corrupted, delete it + if let Err(del_err) = store.del(&key) { + tracing::error!("Failed to delete a corrupted event: {}", del_err); + } } } } @@ -146,11 +171,20 @@ impl WebhookAdapter { /// Send a single HTTP request async fn send_request(&self, event: &Event) -> Result<(), Error> { // Send a request - let response = self.build_request(event).send().await?; + let response = self + .build_request(event) + .send() + .await + .map_err(|e| Error::Custom(format!("Sending a webhook request failed:{}", e)))?; // Check the response status if !response.status().is_success() { - return Err(Error::Custom(format!("Webhook request failed, status code:{}", response.status()))); + let status = response.status(); + let body = response + .text() + .await + .unwrap_or_else(|_| "Unable to read response body".to_string()); + return Err(Error::Custom(format!("Webhook request failed, status code:{},response:{}", status, body))); } Ok(()) @@ -172,18 +206,32 @@ impl WebhookAdapter { } } if let Some(headers) = &self.config.custom_headers { + let mut header_map = HeaderMap::new(); for (key, value) in headers { - request = request.header(key, value); + if let (Ok(name), Ok(val)) = (HeaderName::from_bytes(key.as_bytes()), HeaderValue::from_str(value)) { + header_map.insert(name, val); + } } + request = request.headers(header_map); } request } + + /// Save the event to the queue + async fn save_to_queue(&self, event: &Event) -> Result<(), Error> { + if let Some(store) = &self.store { + store + .put(event.clone()) + .map_err(|e| Error::Custom(format!("Saving events to queue failed: {}", e)))?; + } + Ok(()) + } } #[async_trait] impl ChannelAdapter for WebhookAdapter { fn name(&self) -> String { - ChannelAdapterType::Webhook.to_string() + Webhook.to_string() } async fn send(&self, event: &Event) -> Result<(), Error> { @@ -191,6 +239,17 @@ impl ChannelAdapter for WebhookAdapter { let _ = self.process_backlog().await; // Send the current event - self.send_with_retry(event).await + match self.send_with_retry(event).await { + Ok(_) => Ok(()), + Err(e) => { + // If the send fails and the queue is enabled, save to the queue + if let Some(_) = &self.store { + tracing::warn!("Failed to send the event and saved to the queue: {}", e); + self.save_to_queue(event).await?; + return Ok(()); + } + Err(e) + } + } } } diff --git a/crates/event/src/config.rs b/crates/event/src/config.rs index 77d35539c..498a6b3a0 100644 --- a/crates/event/src/config.rs +++ b/crates/event/src/config.rs @@ -9,7 +9,7 @@ use tracing::info; const DEFAULT_CONFIG_FILE: &str = "event"; /// The prefix for the configuration file -pub(crate) const STORE_PREFIX: &str = "rustfs"; +pub const STORE_PREFIX: &str = "rustfs"; /// The default retry interval for the webhook adapter pub const DEFAULT_RETRY_INTERVAL: u64 = 3; @@ -371,7 +371,7 @@ fn default_queue_dir() -> String { } /// Provides the recommended default channel capacity for high concurrency systems -fn default_queue_limit() -> u64 { +pub(crate) fn default_queue_limit() -> u64 { env::var("EVENT_CHANNEL_CAPACITY") .unwrap_or_else(|_| "10000".to_string()) .parse() @@ -404,7 +404,7 @@ impl EventNotifierConfig { // The existing implementation remains the same, but returns EventNotifierConfig // ... - EventNotifierConfig::default() + Self::default() } /// Deserialization configuration diff --git a/crates/event/src/lib.rs b/crates/event/src/lib.rs index 16350e9c9..cecddecb2 100644 --- a/crates/event/src/lib.rs +++ b/crates/event/src/lib.rs @@ -33,9 +33,4 @@ pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Obje pub use global::{initialize, is_initialized, is_ready, send_event, shutdown}; pub use notifier::NotifierSystem; pub use store::event::EventStore; - -pub use store::get_event_notifier_config; pub use store::queue::QueueStore; - -pub use store::EventSys; -pub use store::GLOBAL_EventSys; diff --git a/crates/event/src/store/manager.rs b/crates/event/src/store/manager.rs index 52fe11956..9f309be2a 100644 --- a/crates/event/src/store/manager.rs +++ b/crates/event/src/store/manager.rs @@ -1,5 +1,4 @@ -use crate::store::{CONFIG_FILE, EVENT}; -use crate::{adapter, ChannelAdapter, EventNotifierConfig, WebhookAdapter}; +use crate::{adapter, ChannelAdapter, EventNotifierConfig}; use common::error::{Error, Result}; use ecstore::config::com::{read_config, save_config, CONFIG_PREFIX}; use ecstore::disk::RUSTFS_META_BUCKET; @@ -12,6 +11,12 @@ use std::sync::Arc; use tokio::sync::Mutex; use tracing::instrument; +/// * config file +const CONFIG_FILE: &str = "event.json"; + +/// event sys config +const EVENT: &str = "event"; + /// Global storage API access point pub static GLOBAL_STORE_API: Lazy>>> = Lazy::new(|| Mutex::new(None)); @@ -44,8 +49,11 @@ impl EventManager { pub async fn init(&self) -> Result { tracing::info!("Event system configuration initialization begins"); - let cfg = match read_event_config(self.api.clone()).await { - Ok(cfg) => cfg, + let cfg = match read_config_without_migrate(self.api.clone()).await { + Ok(cfg) => { + tracing::info!("The event system configuration was successfully read"); + cfg + } Err(err) => { tracing::error!("Failed to initialize the event system configuration:{:?}", err); return Err(err); @@ -53,6 +61,7 @@ impl EventManager { }; *GLOBAL_EVENT_CONFIG.lock().await = Some(cfg.clone()); + tracing::info!("The initialization of the event system configuration is complete"); Ok(cfg) @@ -152,68 +161,29 @@ impl EventManager { None => return Err(Error::msg("The global configuration is not initialized")), }; - let mut adapters: Vec> = Vec::new(); - - // Create a webhook adapter - for (_, webhook_config) in &config.webhook { - if webhook_config.common.enable { - #[cfg(feature = "webhook")] - { - adapters.push(Arc::new(WebhookAdapter::new(webhook_config.clone()))); - } - - #[cfg(not(feature = "webhook"))] - { - return Err(Error::msg("The webhook feature is not enabled")); - } + let adapter_configs = config.to_adapter_configs(); + match adapter::create_adapters(&adapter_configs) { + Ok(adapters) => Ok(adapters), + Err(err) => { + tracing::error!("Failed to create adapters: {:?}", err); + Err(Error::from(err)) } } - - // Create a Kafka adapter - for (_, kafka_config) in &config.kafka { - if kafka_config.common.enable { - #[cfg(all(feature = "kafka", target_os = "linux"))] - { - match KafkaAdapter::new(kafka_config.clone()) { - Ok(adapter) => adapters.push(Arc::new(adapter)), - Err(e) => tracing::error!("Failed to create a Kafka adapter:{}", e), - } - } - - #[cfg(any(not(feature = "kafka"), not(target_os = "linux")))] - { - return Err(Error::msg("Kafka functionality is not enabled or is not a Linux environment")); - } - } - } - - // Create an MQTT adapter - for (_, mqtt_config) in &config.mqtt { - if mqtt_config.common.enable { - #[cfg(feature = "mqtt")] - { - // Implement MQTT adapter creation logic - // ... - } - - #[cfg(not(feature = "mqtt"))] - { - return Err(Error::msg("MQTT The feature is not enabled")); - } - } - } - - Ok(adapters) } } +/// Get the Global Storage API +pub async fn get_global_store_api() -> Option> { + GLOBAL_STORE_API.lock().await.clone() +} + /// Get the Global Storage API pub async fn get_global_event_config() -> Option { GLOBAL_EVENT_CONFIG.lock().await.clone() } /// Read event configuration -async fn read_event_config(api: Arc) -> Result { +async fn read_event_config(api: Arc) -> Result { let config_file = get_event_config_file(); let data = read_config(api, &config_file).await?; @@ -221,7 +191,7 @@ async fn read_event_config(api: Arc) -> Result { } /// Save the event configuration -async fn save_event_config(api: Arc, config: &EventNotifierConfig) -> Result<()> { +async fn save_event_config(api: Arc, config: &EventNotifierConfig) -> Result<()> { let config_file = get_event_config_file(); let data = config.marshal()?; @@ -231,6 +201,38 @@ async fn save_event_config(api: Arc, config: &EventNotifierConfig) -> R /// Get the event profile path fn get_event_config_file() -> String { - // "event/config.json".to_string() format!("{}{}{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, EVENT, SLASH_SEPARATOR, CONFIG_FILE) } + +/// Read the configuration file and create a default configuration if it doesn't exist +pub async fn read_config_without_migrate(api: Arc) -> Result { + let config_file = get_event_config_file(); + let data = match read_config(api.clone(), &config_file).await { + Ok(data) => { + if data.is_empty() { + return new_and_save_event_config(api).await; + } + data + } + Err(err) if ecstore::config::error::is_err_config_not_found(&err) => { + tracing::warn!("If the configuration file does not exist, start initializing the default configuration"); + return new_and_save_event_config(api).await; + } + Err(err) => { + tracing::error!("Read configuration file error: {:?}", err); + return Err(err); + } + }; + + // Parse configuration + let cfg = EventNotifierConfig::unmarshal(&data)?; + Ok(cfg) +} + +/// Create and save a new configuration +async fn new_and_save_event_config(api: Arc) -> Result { + let cfg = EventNotifierConfig::default(); + save_event_config(api, &cfg).await?; + + Ok(cfg) +} diff --git a/crates/event/src/store/mod.rs b/crates/event/src/store/mod.rs index be58ec0ff..6a90338e4 100644 --- a/crates/event/src/store/mod.rs +++ b/crates/event/src/store/mod.rs @@ -1,121 +1,3 @@ pub(crate) mod event; mod manager; pub(crate) mod queue; - -use crate::NotifierConfig; -use common::error::Result; -use ecstore::config::com::{read_config, save_config, CONFIG_PREFIX}; -use ecstore::config::error::is_err_config_not_found; -use ecstore::store::ECStore; -use ecstore::utils::path::SLASH_SEPARATOR; -use ecstore::StorageAPI; -use lazy_static::lazy_static; -use std::sync::{Arc, OnceLock}; -use tracing::{error, instrument, warn}; - -lazy_static! { - pub static ref GLOBAL_EventSys: EventSys = EventSys::new(); - pub static ref GLOBAL_EventSysConfig: OnceLock = OnceLock::new(); -} -/// * config file -const CONFIG_FILE: &str = "event.json"; - -/// event sys config -const EVENT: &str = "event"; - -#[derive(Debug)] -pub struct EventSys {} - -impl Default for EventSys { - fn default() -> Self { - Self::new() - } -} - -impl EventSys { - pub fn new() -> Self { - Self {} - } - #[instrument(skip_all)] - pub async fn init(&self, api: Arc) -> Result<()> { - tracing::info!("event sys config init start"); - let cfg = read_config_without_migrate(api.clone().clone()).await?; - let _ = GLOBAL_EventSysConfig.set(cfg); - tracing::info!("event sys config init done"); - Ok(()) - } -} - -/// get event sys config file -/// -/// # Returns -/// NotifierConfig -pub fn get_event_notifier_config() -> &'static NotifierConfig { - GLOBAL_EventSysConfig.get_or_init(NotifierConfig::default) -} - -fn get_event_sys_file() -> String { - format!("{}{}{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, EVENT, SLASH_SEPARATOR, CONFIG_FILE) -} - -/// new server config -/// -/// # Returns -/// NotifierConfig -fn new_server_config() -> NotifierConfig { - NotifierConfig::new() -} - -async fn new_and_save_server_config(api: Arc) -> Result { - let cfg = new_server_config(); - save_server_config(api, &cfg).await?; - - Ok(cfg) -} - -/// save server config -/// -/// # Parameters -/// - `api`: StorageAPI -/// - `cfg`: configuration to save -/// -/// # Returns -/// Result -async fn save_server_config(api: Arc, cfg: &NotifierConfig) -> Result<()> { - let data = cfg.marshal()?; - - let config_file = get_event_sys_file(); - - save_config(api, &config_file, data).await -} - -/// read config without migrate -/// -/// # Parameters -/// - `api`: StorageAPI -/// -/// # Returns -/// Configuration information -pub async fn read_config_without_migrate(api: Arc) -> Result { - let config_file = get_event_sys_file(); - let data = match read_config(api.clone(), config_file.as_str()).await { - Ok(data) => { - if data.is_empty() { - return new_and_save_server_config(api).await; - } - data - } - Err(err) if is_err_config_not_found(&err) => { - warn!("config not found, start to init"); - return new_and_save_server_config(api).await; - } - Err(err) => { - error!("read config error: {:?}", err); - return Err(err); - } - }; - - // TODO: decrypt if needed - let cfg = NotifierConfig::unmarshal(data.as_slice())?; - Ok(cfg.merge()) -} diff --git a/crates/event/src/store/queue.rs b/crates/event/src/store/queue.rs index bde850748..e7c859531 100644 --- a/crates/event/src/store/queue.rs +++ b/crates/event/src/store/queue.rs @@ -192,7 +192,7 @@ where /// Write the object to a file fn write_object(&self, key: &Key, item: &T) -> Result<()> { - // 序列化对象 + // Serialize the object let data = serde_json::to_vec(item)?; self.write_bytes(key, &data) } @@ -213,7 +213,7 @@ where let path = self.directory.join(key.to_string()); let file_data = if key.compress { - // 使用 snap 压缩数据 + // Use snap to compress data let mut encoder = Encoder::new(); encoder .compress_vec(data) @@ -271,7 +271,10 @@ where // Read existing files let files = self.list_files()?; - let mut entries = self.entries.write().map_err(|_| Error::msg("获取写锁失败"))?; + let mut entries = self + .entries + .write() + .map_err(|_| Error::msg("Failed to obtain a write lock"))?; for file in files { if let Ok(meta) = file.metadata() { @@ -293,7 +296,7 @@ where fn put(&self, item: T) -> Result { { - let entries = self.entries.read().map_err(|_| Error::msg("获取读锁失败"))?; + let entries = self.entries.read().map_err(|_| Error::msg("Failed to obtain a read lock"))?; if entries.len() as u64 >= self.entry_limit { return Err(Error::msg("The storage limit has been reached")); } @@ -314,7 +317,7 @@ where } { - let entries = self.entries.read().map_err(|_| Error::msg("获取读锁失败"))?; + let entries = self.entries.read().map_err(|_| Error::msg("Failed to obtain a read lock"))?; if entries.len() as u64 >= self.entry_limit { return Err(Error::msg("The storage limit has been reached")); } @@ -397,7 +400,7 @@ where fn put_raw(&self, data: &[u8]) -> Result { { - let entries = self.entries.read().map_err(|_| Error::msg("获取读锁失败"))?; + let entries = self.entries.read().map_err(|_| Error::msg("Failed to obtain a read lock"))?; if entries.len() as u64 >= self.entry_limit { return Err(Error::msg("the storage limit has been reached")); } @@ -441,7 +444,10 @@ where } // Remove the entry from the map - let mut entries = self.entries.write().map_err(|_| Error::msg("获取写锁失败"))?; + let mut entries = self + .entries + .write() + .map_err(|_| Error::msg("Failed to obtain a write lock"))?; entries.remove(&key.to_string()); Ok(()) diff --git a/ecstore/src/file_meta.rs b/ecstore/src/file_meta.rs index cf9052b48..5a4f48aee 100644 --- a/ecstore/src/file_meta.rs +++ b/ecstore/src/file_meta.rs @@ -2520,7 +2520,7 @@ mod test { fi3.mod_time = Some(time3); fm.add_version(fi3).unwrap(); - // Sort first to ensure latest is at the front + // Sort first to ensure latest is at the front fm.sort_by_mod_time(); // Should return the first version's mod time (lastest_mod_time returns first version's time) @@ -2690,7 +2690,7 @@ mod test { assert!(result.is_err()); } - #[test] + #[test] fn test_is_latest_delete_marker() { // Test the is_latest_delete_marker function with simple data // Since the function is complex and requires specific XL format, @@ -2798,9 +2798,7 @@ async fn test_file_info_from_raw() { let encoded = fm.marshal_msg().unwrap(); - let raw_info = RawFileInfo { - buf: encoded, - }; + let raw_info = RawFileInfo { buf: encoded }; let result = file_info_from_raw(raw_info, "test-bucket", "test-object", false).await; assert!(result.is_ok()); @@ -2833,26 +2831,26 @@ fn test_file_meta_load_function() { assert!(result.is_err()); } - #[test] - fn test_file_meta_read_bytes_header() { - // Test read_bytes_header function - it expects the first 5 bytes to be msgpack bin length - // Create a buffer with proper msgpack bin format for a 9-byte binary - let mut buf = vec![0xc4, 0x09]; // msgpack bin8 format for 9 bytes - buf.extend_from_slice(b"test data"); // 9 bytes of data - buf.extend_from_slice(b"extra"); // additional data +#[test] +fn test_file_meta_read_bytes_header() { + // Test read_bytes_header function - it expects the first 5 bytes to be msgpack bin length + // Create a buffer with proper msgpack bin format for a 9-byte binary + let mut buf = vec![0xc4, 0x09]; // msgpack bin8 format for 9 bytes + buf.extend_from_slice(b"test data"); // 9 bytes of data + buf.extend_from_slice(b"extra"); // additional data - let result = FileMeta::read_bytes_header(&buf); - assert!(result.is_ok()); - let (length, remaining) = result.unwrap(); - assert_eq!(length, 9); // "test data" length - // remaining should be everything after the 5-byte header (but we only have 2-byte header) - assert_eq!(remaining.len(), buf.len() - 5); + let result = FileMeta::read_bytes_header(&buf); + assert!(result.is_ok()); + let (length, remaining) = result.unwrap(); + assert_eq!(length, 9); // "test data" length + // remaining should be everything after the 5-byte header (but we only have 2-byte header) + assert_eq!(remaining.len(), buf.len() - 5); - // Test with buffer too small - let small_buf = vec![0u8; 2]; - let result = FileMeta::read_bytes_header(&small_buf); - assert!(result.is_err()); - } + // Test with buffer too small + let small_buf = vec![0u8; 2]; + let result = FileMeta::read_bytes_header(&small_buf); + assert!(result.is_err()); +} #[test] fn test_file_meta_get_set_idx() { @@ -3080,11 +3078,11 @@ fn test_file_meta_version_header_ordering() { // Test partial_cmp assert!(header1.partial_cmp(&header2).is_some()); - // Test cmp - header2 should be greater (newer) - use std::cmp::Ordering; - assert_eq!(header1.cmp(&header2), Ordering::Less); // header1 has earlier time - assert_eq!(header2.cmp(&header1), Ordering::Greater); // header2 has later time - assert_eq!(header1.cmp(&header1), Ordering::Equal); + // Test cmp - header2 should be greater (newer) + use std::cmp::Ordering; + assert_eq!(header1.cmp(&header2), Ordering::Less); // header1 has earlier time + assert_eq!(header2.cmp(&header1), Ordering::Greater); // header2 has later time + assert_eq!(header1.cmp(&header1), Ordering::Equal); } #[test] @@ -3110,10 +3108,7 @@ fn test_merge_file_meta_versions_edge_cases() { version2.header.version_id = Some(Uuid::new_v4()); version2.header.mod_time = Some(OffsetDateTime::from_unix_timestamp(2000).unwrap()); - let versions = vec![ - vec![version1.clone()], - vec![version2.clone()], - ]; + let versions = vec![vec![version1.clone()], vec![version2.clone()]]; let _merged_strict = merge_file_meta_versions(1, true, 10, &versions); let merged_non_strict = merge_file_meta_versions(1, false, 10, &versions); @@ -3191,9 +3186,7 @@ async fn test_get_file_info_edge_cases() { #[tokio::test] async fn test_file_info_from_raw_edge_cases() { // Test with empty buffer - let empty_raw = RawFileInfo { - buf: vec![], - }; + let empty_raw = RawFileInfo { buf: vec![] }; let result = file_info_from_raw(empty_raw, "bucket", "object", false).await; assert!(result.is_err()); @@ -3227,12 +3220,12 @@ fn test_meta_object_edge_cases() { obj.data_dir = None; assert!(obj.use_data_dir()); - // Test use_inlinedata (always returns false in current implementation) - obj.size = 128 * 1024; // 128KB threshold - assert!(!obj.use_inlinedata()); // Should be false + // Test use_inlinedata (always returns false in current implementation) + obj.size = 128 * 1024; // 128KB threshold + assert!(!obj.use_inlinedata()); // Should be false - obj.size = 128 * 1024 - 1; - assert!(!obj.use_inlinedata()); // Should also be false (always false) + obj.size = 128 * 1024 - 1; + assert!(!obj.use_inlinedata()); // Should also be false (always false) } #[test] @@ -3244,17 +3237,17 @@ fn test_file_meta_version_header_edge_cases() { header.ec_m = 0; assert!(!header.has_ec()); - // Test matches_not_strict with different signatures but same version_id - let mut other = FileMetaVersionHeader::default(); - let version_id = Some(Uuid::new_v4()); - header.version_id = version_id; - other.version_id = version_id; - header.version_type = VersionType::Object; - other.version_type = VersionType::Object; - header.signature = [1, 2, 3, 4]; - other.signature = [5, 6, 7, 8]; - // Should match because they have same version_id and type - assert!(header.matches_not_strict(&other)); + // Test matches_not_strict with different signatures but same version_id + let mut other = FileMetaVersionHeader::default(); + let version_id = Some(Uuid::new_v4()); + header.version_id = version_id; + other.version_id = version_id; + header.version_type = VersionType::Object; + other.version_type = VersionType::Object; + header.signature = [1, 2, 3, 4]; + other.signature = [5, 6, 7, 8]; + // Should match because they have same version_id and type + assert!(header.matches_not_strict(&other)); // Test sorts_before with same mod_time but different version_id let time = OffsetDateTime::from_unix_timestamp(1000).unwrap(); @@ -3286,12 +3279,12 @@ fn test_file_meta_add_version_edge_cases() { fi2.mod_time = Some(OffsetDateTime::now_utc()); fm.add_version(fi2).unwrap(); - // Should still have only one version, but updated - assert_eq!(fm.versions.len(), 1); - let (_, version) = fm.find_version(version_id).unwrap(); - if let Some(obj) = version.object { - assert_eq!(obj.size, 2048); // Size gets updated when adding same version_id - } + // Should still have only one version, but updated + assert_eq!(fm.versions.len(), 1); + let (_, version) = fm.find_version(version_id).unwrap(); + if let Some(obj) = version.object { + assert_eq!(obj.size, 2048); // Size gets updated when adding same version_id + } } #[test] @@ -3324,12 +3317,11 @@ fn test_file_meta_shard_data_dir_count_edge_cases() { fi.mod_time = Some(OffsetDateTime::now_utc()); fm.add_version(fi).unwrap(); - let count = fm.shard_data_dir_count(&version_id, &data_dir); - assert_eq!(count, 0); // Should be 0 because user_data_dir() requires flag + let count = fm.shard_data_dir_count(&version_id, &data_dir); + assert_eq!(count, 0); // Should be 0 because user_data_dir() requires flag // Test with different version_id let other_version_id = Some(Uuid::new_v4()); - let count = fm.shard_data_dir_count(&other_version_id, &data_dir); - assert_eq!(count, 1); // Should be 1 because the version has matching data_dir and user_data_dir() is true + let count = fm.shard_data_dir_count(&other_version_id, &data_dir); + assert_eq!(count, 1); // Should be 1 because the version has matching data_dir and user_data_dir() is true } - diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index 68d78ee2d..2c0bf0d23 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -5890,7 +5890,7 @@ mod tests { erasure: ErasureInfo { data_blocks: 4, parity_blocks: 2, - index: 1, // Must be > 0 for is_valid() to return true + index: 1, // Must be > 0 for is_valid() to return true distribution: vec![1, 2, 3, 4, 5, 6], // Must match data_blocks + parity_blocks ..Default::default() }, @@ -5902,7 +5902,7 @@ mod tests { erasure: ErasureInfo { data_blocks: 6, parity_blocks: 3, - index: 1, // Must be > 0 for is_valid() to return true + index: 1, // Must be > 0 for is_valid() to return true distribution: vec![1, 2, 3, 4, 5, 6, 7, 8, 9], // Must match data_blocks + parity_blocks ..Default::default() }, @@ -5914,7 +5914,7 @@ mod tests { erasure: ErasureInfo { data_blocks: 2, parity_blocks: 1, - index: 1, // Must be > 0 for is_valid() to return true + index: 1, // Must be > 0 for is_valid() to return true distribution: vec![1, 2, 3], // Must match data_blocks + parity_blocks ..Default::default() }, @@ -6019,11 +6019,7 @@ mod tests { #[test] fn test_join_errs() { // Test joining error messages - let errs = vec![ - None, - Some(Error::from_string("error1")), - Some(Error::from_string("error2")), - ]; + let errs = vec![None, Some(Error::from_string("error1")), Some(Error::from_string("error2"))]; let joined = join_errs(&errs); assert!(joined.contains("")); assert!(joined.contains("error1")); diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 8c8f4f004..4d749270c 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -1153,7 +1153,11 @@ mod tests { assert_eq!(file_info.get_etag(), None); // With etag - file_info.metadata.as_mut().unwrap().insert("etag".to_string(), "test-etag".to_string()); + file_info + .metadata + .as_mut() + .unwrap() + .insert("etag".to_string(), "test-etag".to_string()); assert_eq!(file_info.get_etag(), Some("test-etag".to_string())); } @@ -1282,10 +1286,7 @@ mod tests { file_info.set_healing(); assert!(file_info.metadata.is_some()); - assert_eq!( - file_info.metadata.as_ref().unwrap().get(RUSTFS_HEALING), - Some(&"true".to_string()) - ); + assert_eq!(file_info.metadata.as_ref().unwrap().get(RUSTFS_HEALING), Some(&"true".to_string())); } #[test] @@ -1656,10 +1657,11 @@ mod tests { assert!(!object_info.is_compressed()); // With compression metadata - object_info.user_defined.as_mut().unwrap().insert( - format!("{}compression", RESERVED_METADATA_PREFIX), - "gzip".to_string() - ); + object_info + .user_defined + .as_mut() + .unwrap() + .insert(format!("{}compression", RESERVED_METADATA_PREFIX), "gzip".to_string()); assert!(object_info.is_compressed()); } @@ -1866,7 +1868,7 @@ mod tests { let shard_size = erasure.shard_size(1000); assert_eq!(shard_size, 1000); // 1000 / 1 = 1000 - // Test with zero block size - this will cause division by zero in shard_size + // Test with zero block size - this will cause division by zero in shard_size // So we need to test with non-zero block_size but zero data_blocks was already fixed above let erasure = ErasureInfo { data_blocks: 4,