fix: clippy and coderabbit suggestions

Author: parmesant

src/cli.rs

@@ -16,7 +16,7 @@
  *
  */
 
-use clap::Parser;
+use clap::{Parser, value_parser};
 use std::{env, fs, path::PathBuf};
 
 use url::Url;
@@ -170,10 +170,11 @@ pub struct Options {
     #[arg(
         long,
         env = "P_ACTIX_NUM_WORKERS",
-        default_value_t = num_cpus::get(),
+        default_value_t = num_cpus::get() as u64,
+        value_parser = value_parser!(u64).range(1..),
         help = "Number of workers for actix-web"
     )]
-    pub num_workers: usize,
+    pub num_workers: u64,
 
     // Actix connections backlog
     #[arg(

src/handlers/http/modal/mod.rs

@@ -21,6 +21,7 @@ use std::{fmt, path::Path, sync::Arc, time::Duration};
 use actix_web::{App, HttpServer, http::KeepAlive, middleware::from_fn, web::ServiceConfig};
 use actix_web_prometheus::PrometheusMetrics;
 use anyhow::Context;
+use arrow::datatypes::ArrowNativeType;
 use async_trait::async_trait;
 use base64::{Engine, prelude::BASE64_STANDARD};
 use bytes::Bytes;
@@ -121,7 +122,7 @@ pub trait ParseableServer {
 
         // Create the HTTP server
         let http_server = HttpServer::new(create_app_fn)
-            .workers(PARSEABLE.options.num_workers)
+            .workers(PARSEABLE.options.num_workers.as_usize())
             .keep_alive(KeepAlive::Timeout(Duration::from_secs(
                 PARSEABLE.options.keep_alive,
             )))

src/handlers/http/modal/utils/ingest_utils.rs

@@ -182,6 +182,12 @@ pub fn push_logs(
         log_source,
     )?;
 
+    if data.is_empty() {
+        return Err(PostError::Invalid(anyhow::Error::msg(
+            "Empty data object received",
+        )));
+    }
+
     // Batch path: one schema inference + one decoder for the entire batch.
     // When custom partitions are set, different records may need different
     // partition keys, so fall back to per-record processing.
@@ -243,7 +249,7 @@ pub fn push_logs(
             }
         });
         if let ControlFlow::Break(e) = r {
-            return Err(PostError::CustomError(e.to_string()));
+            return Err(PostError::Invalid(e));
         }
     }
 

src/otel/metrics.rs

@@ -460,6 +460,23 @@ pub fn flatten_metrics_record(metrics_record: &Metric) -> Vec<Map<String, Value>
         single.insert("metric_description".to_string(), metric_desc);
         single.insert("metric_unit".to_string(), metric_unit);
         single.insert("metric_type".to_string(), metric_type_val);
+        match &metrics_record.data {
+            Some(metric::Data::Sum(sum)) => {
+                single.extend(flatten_aggregation_temporality(sum.aggregation_temporality));
+                single.insert("is_monotonic".to_string(), Value::Bool(sum.is_monotonic));
+            }
+            Some(metric::Data::Histogram(histogram)) => {
+                single.extend(flatten_aggregation_temporality(
+                    histogram.aggregation_temporality,
+                ));
+            }
+            Some(metric::Data::ExponentialHistogram(exp_histogram)) => {
+                single.extend(flatten_aggregation_temporality(
+                    exp_histogram.aggregation_temporality,
+                ));
+            }
+            _ => {}
+        }
         single.extend(metadata);
         return vec![single];
     }

src/parseable/staging/reader.rs

@@ -143,7 +143,7 @@ pub struct OffsetReader<R: Read + Seek> {
 }
 
 impl<R: Read + Seek> OffsetReader<R> {
-    fn _new(reader: R, offset_list: Vec<(u64, usize)>) -> Self {
+    fn new(reader: R, offset_list: Vec<(u64, usize)>) -> Self {
         let mut offset_list = offset_list.into_iter();
         let mut finished = false;
 
@@ -217,7 +217,7 @@ pub fn get_reverse_reader<T: Read + Seek>(
     let mut offset = 0;
     let mut messages = Vec::new();
 
-    while let Some(res) = _find_limit_and_type(&mut reader).transpose() {
+    while let Some(res) = find_limit_and_type(&mut reader).transpose() {
         match res {
             Ok((header, size)) => {
                 messages.push((header, offset, size));
@@ -238,11 +238,16 @@ pub fn get_reverse_reader<T: Read + Seek>(
     // reset reader
     reader.rewind()?;
 
-    Ok(StreamReader::try_new(BufReader::new(OffsetReader::_new(reader, messages)), None).unwrap())
+    StreamReader::try_new(BufReader::new(OffsetReader::new(reader, messages)), None).map_err(|e| {
+        io::Error::new(
+            io::ErrorKind::InvalidData,
+            format!("Invalid arrow stream: {e}"),
+        )
+    })
 }
 
 // return limit for
-fn _find_limit_and_type(
+fn find_limit_and_type(
     reader: &mut (impl Read + Seek),
 ) -> Result<Option<(MessageHeader, usize)>, io::Error> {
     let mut size = 0;
@@ -476,7 +481,7 @@ mod tests {
         // Define offset list: (offset, size)
         let offsets = vec![(2, 3), (7, 2)]; // Read bytes 2-4 (3, 4, 5) and then 7-8 (8, 9)
 
-        let mut reader = OffsetReader::_new(cursor, offsets);
+        let mut reader = OffsetReader::new(cursor, offsets);
         let mut buffer = [0u8; 10];
 
         // First read should get bytes 3, 4, 5
@@ -563,7 +568,7 @@ mod tests {
         let data = vec![1, 2, 3, 4, 5];
         let cursor = Cursor::new(data);
 
-        let mut reader = OffsetReader::_new(cursor, vec![]);
+        let mut reader = OffsetReader::new(cursor, vec![]);
         let mut buffer = [0u8; 10];
 
         // Should return 0 bytes read
@@ -580,7 +585,7 @@ mod tests {
         // One offset of 5 bytes
         let offsets = vec![(2, 5)]; // Read bytes 2-6 (3, 4, 5, 6, 7)
 
-        let mut reader = OffsetReader::_new(cursor, offsets);
+        let mut reader = OffsetReader::new(cursor, offsets);
         let mut buffer = [0u8; 3]; // Buffer smaller than the 5 bytes we want to read
 
         // First read should get first 3 bytes: 3, 4, 5
@@ -641,7 +646,7 @@ mod tests {
         // One large offset (8KB)
         let offsets = vec![(1000, 8000)];
 
-        let mut reader = OffsetReader::_new(cursor, offsets);
+        let mut reader = OffsetReader::new(cursor, offsets);
         let mut buffer = [0u8; 10000];
 
         // Should read 8KB

src/sync.rs

@@ -31,6 +31,16 @@ use tracing::{Instrument, error, info, info_span, trace, warn};
 static LOCAL_SYNC_RUNNING: AtomicBool = AtomicBool::new(false);
 static REMOTE_SYNC_RUNNING: AtomicBool = AtomicBool::new(false);
 
+/// RAII guard that clears a sync-running flag on drop, so a panic inside the
+/// sync body cannot leave the flag stuck at `true` and wedge future ticks.
+struct SyncRunningGuard(&'static AtomicBool);
+
+impl Drop for SyncRunningGuard {
+    fn drop(&mut self) {
+        self.0.store(false, Ordering::SeqCst);
+    }
+}
+
 use crate::alerts::alert_enums::AlertTask;
 use crate::alerts::alerts_utils;
 use crate::parseable::PARSEABLE;
@@ -135,6 +145,7 @@ pub fn object_store_sync() -> (
                             warn!("Previous object_store_sync cycle still running, skipping this tick");
                             continue;
                         }
+                        let _guard = SyncRunningGuard(&REMOTE_SYNC_RUNNING);
                         async {
                             trace!("Syncing Parquets to Object Store... ");
 
@@ -153,7 +164,6 @@ pub fn object_store_sync() -> (
                                 }
                             ).await;
                         }.instrument(info_span!("object_store_sync_cycle")).await;
-                        REMOTE_SYNC_RUNNING.store(false, Ordering::SeqCst);
                     },
                     res = &mut inbox_rx => {
                         match res {
@@ -209,6 +219,7 @@ pub fn local_sync() -> (
                             warn!("Previous local_sync cycle still running, skipping this tick");
                             continue;
                         }
+                        let _guard = SyncRunningGuard(&LOCAL_SYNC_RUNNING);
                         // Monitor the duration of flush_and_convert execution
                         async {
                             monitor_task_duration(
@@ -225,7 +236,6 @@ pub fn local_sync() -> (
                                 }
                             ).await;
                         }.instrument(info_span!("local_sync_cycle")).await;
-                        LOCAL_SYNC_RUNNING.store(false, Ordering::SeqCst);
                     },
                     res = &mut inbox_rx => {
                         match res {

src/telemetry.rs

@@ -26,7 +26,9 @@ use opentelemetry_sdk::{
 };
 // Consts describing the env vars
 const OTEL_EXPORTER_OTLP_ENDPOINT: &str = "OTEL_EXPORTER_OTLP_ENDPOINT";
+const OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: &str = "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT";
 const OTEL_EXPORTER_OTLP_PROTOCOL: &str = "OTEL_EXPORTER_OTLP_PROTOCOL";
+
 /// Initialise an OTLP tracer provider.
 ///
 /// **Required env var:**
@@ -45,12 +47,16 @@ const OTEL_EXPORTER_OTLP_PROTOCOL: &str = "OTEL_EXPORTER_OTLP_PROTOCOL";
 ///   as gRPC metadata or HTTP headers, e.g.
 ///   \`authorization=Basic <token>,x-p-stream=my-stream,x-p-log-source=otel-traces\`
 ///
-/// Returns \`None\` when \`OTEL_EXPORTER_OTLP_ENDPOINT\` is not set (OTEL disabled).
+/// Returns \`None\` when \`OTEL_EXPORTER_OTLP_ENDPOINT\` or `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` is not set (OTEL disabled).
 /// The caller must call \`provider.shutdown()\` before process exit.
 pub fn init_tracing() -> Option<SdkTracerProvider> {
     // Only used to decide whether OTEL is enabled; the SDK reads it again
     // from env to build the exporter (which also appends /v1/traces for HTTP).
-    std::env::var(OTEL_EXPORTER_OTLP_ENDPOINT).ok()?;
+    if std::env::var(OTEL_EXPORTER_OTLP_ENDPOINT).is_err()
+        && std::env::var(OTEL_EXPORTER_OTLP_TRACES_ENDPOINT).is_err()
+    {
+        return None;
+    }
 
     let protocol =
         std::env::var(OTEL_EXPORTER_OTLP_PROTOCOL).unwrap_or_else(|_| "http/json".to_string());