fix: deadlock in metric pruning

in case schema hashes grow beyond NUM_CPU, deadlock can occur during sort and prune

Author: parmesant

src/handlers/http/ingest.rs

@@ -24,6 +24,8 @@ use actix_web::{HttpRequest, HttpResponse, http::header::ContentType};
 use arrow_array::RecordBatch;
 use bytes::Bytes;
 use chrono::Utc;
+use once_cell::sync::Lazy;
+use rayon::{ThreadPool, ThreadPoolBuilder};
 use tokio::sync::oneshot;
 use tracing::error;
 
@@ -55,6 +57,9 @@ use super::modal::utils::ingest_utils::{flatten_and_push_logs, get_custom_fields
 use super::users::dashboards::DashboardError;
 use super::users::filters::FiltersError;
 
+pub static INGESTION_THREADPOOL: Lazy<ThreadPool> =
+    Lazy::new(|| ThreadPoolBuilder::new().build().unwrap());
+
 // Handler for POST /api/v1/ingest
 // ingests events by extracting stream name from header
 // creates if stream does not exist

src/parseable/streams.rs

@@ -56,7 +56,7 @@ use crate::{
         DEFAULT_TIMESTAMP_KEY,
         format::{LogSource, LogSourceEntry},
     },
-    handlers::DatasetTag,
+    handlers::{DatasetTag, http::ingest::INGESTION_THREADPOOL},
     hottier::StreamHotTier,
     metadata::{LogStreamMetadata, SchemaVersion},
     metrics,
@@ -757,7 +757,7 @@ impl Stream {
         time_partition_field: String,
     ) -> std::sync::mpsc::Receiver<Result<PreparedMetricRowGroup, StagingError>> {
         let (tx, rx) = std::sync::mpsc::sync_channel(1);
-        rayon::spawn(move || {
+        INGESTION_THREADPOOL.spawn(move || {
             let _ = tx.send(Self::prepare_metric_row_group(
                 schema,
                 buffer,