Refactor unwrap to improve error handling

- Refactor `Mutex<bool>` to `AtomicBool` to prevent potential panics and enhance performance.

src/graphql/export.rs

@@ -2669,8 +2669,9 @@ fn export_statistic_file(
             loop {
                 match iter.next() {
                     Some(Ok(item)) => {
-                        // replace new value (min_index's vector value is always exist)
-                        *iter_next_values.get_mut(min_index).unwrap() = item;
+                        *iter_next_values
+                            .get_mut(min_index)
+                            .expect("The index is one of the actual elements in the vector, so it is always valid.") = item;
                         break;
                     }
                     Some(Err(_)) => {

src/graphql/statistics.rs

@@ -224,8 +224,7 @@ fn gen_statistics(
         for idx in next_candidate {
             if let Some(iter) = stats_iters.get_mut(idx) {
                 if let Some(item) = iter.next() {
-                    // replace new value (value is always exist)
-                    *iter_next_values.get_mut(idx).unwrap() = item;
+                    *iter_next_values.get_mut(idx).expect("`next_candidate` is generated during iteration over `iter_next_values`, ensuring all its indices are valid within the latter.") = item;
                 } else {
                     // No value to call with the iterator.
                     let _ = stats_iters.remove(idx);

src/lib.rs

@@ -165,8 +165,10 @@ extern crate proc_macro;
 /// ```
 #[proc_macro_derive(ConvertGraphQLEdgesNode, attributes(graphql_client_type))]
 pub fn derive_from_graphql_client_autogen(input: TokenStream) -> TokenStream {
-    derive_from_graphql_client_autogen_2(input.into())
-        .unwrap()
+    derive_from_graphql_client_autogen_2(input.clone().into())
+        .unwrap_or_else(|e| {
+            panic!("ConvertGraphQLEdgesNode macro is not correctly used for {input:?}: {e}")
+        })
         .into()
 }
 

src/main.rs

@@ -12,7 +12,10 @@ use std::{
     fs,
     net::SocketAddr,
     path::Path,
-    sync::{Arc, Mutex},
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
     time::Duration,
 };
 
@@ -126,7 +129,7 @@ async fn main() -> Result<()> {
         let notify_shutdown = Arc::new(Notify::new());
         let notify_reboot = Arc::new(Notify::new());
         let notify_power_off = Arc::new(Notify::new());
-        let retain_flag = Arc::new(Mutex::new(false));
+        let retain_flag = Arc::new(AtomicBool::new(false));
 
         let database = if let Some(settings) = settings.clone() {
             let (db_path, db_options) = db_path_and_option(
@@ -312,11 +315,8 @@ async fn main() -> Result<()> {
 
         if is_reboot || is_power_off || is_reload_config {
             loop {
-                {
-                    let retain_flag = retain_flag.lock().unwrap();
-                    if !*retain_flag {
-                        break;
-                    }
+                if !retain_flag.load(Ordering::Relaxed) {
+                    break;
                 }
                 sleep(Duration::from_millis(SERVER_REBOOT_DELAY)).await;
             }

src/peer.rs

@@ -717,8 +717,12 @@ async fn update_to_new_peer_list(
         if let Err(e) = insert_toml_peers(&mut doc, Some(data)) {
             error!("{e:?}");
         }
-        if let Err(e) = write_toml_file(&doc, &path.unwrap()) {
-            error!("{e:?}");
+        if let Some(path) = path {
+            if let Err(e) = write_toml_file(&doc, &path) {
+                error!("{e:?}");
+            }
+        } else {
+            error!("Configuration path is None, unable to write to the TOML file.");
         }
     }
 

src/publish.rs

@@ -528,8 +528,7 @@ where
             info!("start hog's publish stream : {:?}", record_type);
         }
         NodeType::Crusher => {
-            // crusher's policy Id always exists.
-            let id = msg.id().unwrap();
+            let id = msg.id().expect("Crusher always sends RequestCrusherStream with an id, so this value is guaranteed to exist.");
             send_crusher_stream_start_message(&mut sender, id)
                 .await
                 .map_err(|e| anyhow!("Failed to write crusher start message: {}", e))?;

src/publish/implement.rs

@@ -1,6 +1,6 @@
 use std::{net::IpAddr, vec};
 
-use anyhow::{bail, Result};
+use anyhow::{anyhow, bail, Result};
 use giganto_client::publish::stream::{NodeType, RequestCrusherStream, RequestHogStream};
 
 pub trait RequestStreamMessage {
@@ -13,14 +13,17 @@ pub trait RequestStreamMessage {
 
 impl RequestStreamMessage for RequestHogStream {
     fn channel_key(&self, sensor: Option<String>, record_type: &str) -> Result<Vec<String>> {
+        let sensor = sensor.ok_or_else(|| {
+            anyhow!("Failed to generate semi-supervised channel key, sensor is required.")
+        })?;
         if let Some(ref sensor_list) = self.sensor {
             let hog_keys = sensor_list
                 .iter()
                 .map(|target_sensor| {
                     let mut key = String::new();
                     key.push_str(&NodeType::Hog.to_string());
                     key.push('\0');
-                    key.push_str(sensor.as_ref().unwrap());
+                    key.push_str(&sensor);
                     key.push('\0');
                     key.push_str(target_sensor);
                     key.push('\0');

src/storage.rs

@@ -8,7 +8,10 @@ use std::{
     ops::Deref,
     path::{Path, PathBuf},
     process::exit,
-    sync::{Arc, Mutex},
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
     time::{Duration, Instant},
 };
 
@@ -943,7 +946,7 @@ pub async fn retain_periodically(
     retention_period: Duration,
     db: Database,
     notify_shutdown: Arc<Notify>,
-    running_flag: Arc<Mutex<bool>>,
+    running_flag: Arc<AtomicBool>,
 ) -> Result<()> {
     const DEFAULT_FROM_TIMESTAMP_NANOS: i64 = 61_000_000_000;
     const ONE_DAY_TIMESTAMP_NANOS: i64 = 86_400_000_000_000;
@@ -955,10 +958,8 @@ pub async fn retain_periodically(
         select! {
             _ = itv.tick() => {
                 info!("Begin to cleanup the database.");
-                {
-                    let mut running_flag = running_flag.lock().unwrap();
-                    *running_flag = true;
-                }
+                running_flag.store(true, Ordering::Relaxed);
+
                 let now = Utc::now();
                 let mut retention_timestamp = now
                     .timestamp_nanos_opt()
@@ -1037,10 +1038,7 @@ pub async fn retain_periodically(
                     }
                 }
                 info!("Database cleanup completed.");
-                {
-                    let mut running_flag = running_flag.lock().unwrap();
-                    *running_flag = false;
-                }
+                running_flag.store(false, Ordering::Relaxed);
             },
             () = notify_shutdown.notified() => {
                 return Ok(());

src/storage/migration.rs

@@ -196,7 +196,7 @@ where
             continue;
         };
         let new_key = StorageKey::builder()
-            .start_key(&netflow_raw_event.sensor().unwrap())
+            .start_key(&netflow_raw_event.sensor().expect("Netflow source field is always populated during processing, ensuring sensor value exists."))
             .end_key(timestamp)
             .build();
         store.append(&new_key.key(), &value)?;