fix: clippy and coderabbit suggestions

parmesant · parmesant · commit 845ccb265549 · 2026-04-28T11:41:42.000+05:30
diff --git a/src/cli.rs b/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(
diff --git a/src/handlers/http/modal/mod.rs b/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,
             )))
diff --git a/src/handlers/http/modal/utils/ingest_utils.rs b/src/handlers/http/modal/utils/ingest_utils.rs
@@ -24,7 +24,7 @@ use opentelemetry_proto::tonic::{
 };
 use rayon::iter::{IntoParallelIterator, ParallelIterator};
 use serde_json::Value;
-use std::{collections::HashMap, ops::ControlFlow};
+use std::collections::HashMap;
 use tracing::{instrument, warn};
 
 use crate::{
@@ -100,7 +100,7 @@ pub async fn flatten_and_push_logs(
                     Value::Array(records),
                     log_source,
                     p_custom_fields,
-                    time_partition,
+                    None,
                     telemetry_type,
                     tenant_id,
                 )?;
@@ -115,7 +115,7 @@ pub async fn flatten_and_push_logs(
                     Value::Array(records),
                     log_source,
                     p_custom_fields,
-                    time_partition,
+                    None,
                     telemetry_type,
                     tenant_id,
                 )?;
@@ -130,7 +130,7 @@ pub async fn flatten_and_push_logs(
                     Value::Array(records),
                     log_source,
                     p_custom_fields,
-                    time_partition,
+                    None,
                     telemetry_type,
                     tenant_id,
                 )?;
@@ -182,10 +182,16 @@ 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.
-    if custom_partition.is_none() {
+    if custom_partition.is_none() && time_partition.is_none() {
         // process_non_partitioned returns vec![Value::Array([...])], a single
         // element wrapping the whole batch. Unwrap it to avoid double-nesting.
         let json_batch = data.into_iter().next().unwrap();
@@ -219,34 +225,29 @@ pub fn push_logs(
         event.process()?;
     } else {
         // Per-record path for custom-partitioned streams
-        let r = data.into_par_iter().try_for_each(|json| {
-            let origin_size = json_byte_size(&json);
-
-            match (json::Event { json, p_timestamp }).into_event(
-                stream_name.to_owned(),
-                origin_size,
-                &schema,
-                static_schema_flag,
-                custom_partition.as_ref(),
-                time_partition.as_ref(),
-                schema_version,
-                StreamType::UserDefined,
-                p_custom_fields,
-                telemetry_type,
-                tenant_id,
-            ) {
-                Ok(event) => match event.process() {
-                    Ok(_) => ControlFlow::Continue(()),
-                    Err(e) => ControlFlow::Break(e.into()),
-                },
-                Err(e) => ControlFlow::Break(e),
-            }
-        });
-        if let ControlFlow::Break(e) = r {
-            return Err(PostError::CustomError(e.to_string()));
+        let events: Result<Vec<_>, _> = data
+            .into_par_iter()
+            .map(|json| {
+                let origin_size = json_byte_size(&json);
+                (json::Event { json, p_timestamp }).into_event(
+                    stream_name.to_owned(),
+                    origin_size,
+                    &schema,
+                    static_schema_flag,
+                    custom_partition.as_ref(),
+                    time_partition.as_ref(),
+                    schema_version,
+                    StreamType::UserDefined,
+                    p_custom_fields,
+                    telemetry_type,
+                    tenant_id,
+                )
+            })
+            .collect();
+        for event in events.map_err(PostError::Invalid)? {
+            event.process()?;
         }
     }
-
     Ok(())
 }
 
diff --git a/src/otel/metrics.rs b/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];
     }
diff --git a/src/parseable/staging/mod.rs b/src/parseable/staging/mod.rs
@@ -17,6 +17,8 @@
  *
  */
 
+use std::sync::PoisonError;
+
 use crate::{parseable::StreamNotFound, tenants::TenantNotFound};
 
 pub mod reader;
@@ -36,6 +38,6 @@ pub enum StagingError {
     StreamNotFound(#[from] StreamNotFound),
     #[error("{0}")]
     TenantNotFound(#[from] TenantNotFound),
-    // #[error("Metadata Error: {0}")]
-    // Metadata(#[from] MetadataError),
+    #[error("{0}")]
+    PoisonError(#[from] PoisonError<String>),
 }
diff --git a/src/parseable/staging/reader.rs b/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
diff --git a/src/parseable/streams.rs b/src/parseable/streams.rs
@@ -33,7 +33,6 @@ use parquet::{
     schema::types::ColumnPath,
 };
 use relative_path::RelativePathBuf;
-use std::io::BufReader;
 use std::{
     collections::{HashMap, HashSet},
     fs::{self, File, OpenOptions, remove_file, write},
@@ -42,6 +41,7 @@ use std::{
     sync::{Arc, Mutex, RwLock},
     time::{Instant, SystemTime, UNIX_EPOCH},
 };
+use std::{io::BufReader, sync::PoisonError};
 use tokio::task::JoinSet;
 use tracing::{Instrument, error, info, info_span, instrument, trace, warn};
 use ulid::Ulid;
@@ -162,7 +162,11 @@ impl Stream {
                         "Writer lock poisoned while ingesting data for stream {}",
                         self.stream_name
                     );
-                    poisoned.into_inner()
+
+                    return Err(StagingError::PoisonError(PoisonError::new(format!(
+                        "Writer lock poisoned while ingesting data for stream {} - {}",
+                        self.stream_name, poisoned
+                    ))));
                 }
             }
         };
diff --git a/src/sync.rs b/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 {
diff --git a/src/telemetry.rs b/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());
