fix: multiple fixes around system stability under load

1. perform object store sync for all streams in parallel
2. remove restriction of multi threading to utilise all available cores
3. add atomicity in conversion by -
    i. each conversion task processes one minute of arrows
   ii. move arrow files to inprocess folder to maintain atomicity
  iii. add a init sync task to process all pending files
   iv. add tokio sleep of 5 secs in shutdown task to let complete ongoing jobs
    v. remove unwrap of write locks to avoid thread poisoning

Author: nikhilsinhaparseable

src/handlers/http/health_check.rs

@@ -29,12 +29,12 @@ use actix_web::{
 use http::StatusCode;
 use once_cell::sync::Lazy;
 use tokio::{sync::Mutex, task::JoinSet};
-use tracing::{error, info, warn};
+use tracing::{error, info};
 
-use crate::parseable::PARSEABLE;
+use crate::{parseable::PARSEABLE, storage::object_storage::sync_all_streams};
 
 // Create a global variable to store signal status
-static SIGNAL_RECEIVED: Lazy<Arc<Mutex<bool>>> = Lazy::new(|| Arc::new(Mutex::new(false)));
+pub static SIGNAL_RECEIVED: Lazy<Arc<Mutex<bool>>> = Lazy::new(|| Arc::new(Mutex::new(false)));
 
 pub async fn liveness() -> HttpResponse {
     HttpResponse::new(StatusCode::OK)
@@ -60,28 +60,33 @@ pub async fn shutdown() {
     let mut shutdown_flag = SIGNAL_RECEIVED.lock().await;
     *shutdown_flag = true;
 
-    let mut joinset = JoinSet::new();
+    //sleep for 5 secs to allow any ongoing requests to finish
+    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+    let mut local_sync_joinset = JoinSet::new();
 
     // Sync staging
-    PARSEABLE.streams.flush_and_convert(&mut joinset, true);
+    PARSEABLE
+        .streams
+        .flush_and_convert(&mut local_sync_joinset, false, true);
 
-    while let Some(res) = joinset.join_next().await {
+    while let Some(res) = local_sync_joinset.join_next().await {
         match res {
             Ok(Ok(_)) => info!("Successfully converted arrow files to parquet."),
-            Ok(Err(err)) => warn!("Failed to convert arrow files to parquet. {err:?}"),
+            Ok(Err(err)) => error!("Failed to convert arrow files to parquet. {err:?}"),
             Err(err) => error!("Failed to join async task: {err}"),
         }
     }
 
-    if let Err(e) = PARSEABLE
-        .storage
-        .get_object_store()
-        .upload_files_from_staging()
-        .await
-    {
-        warn!("Failed to sync local data with object store. {:?}", e);
-    } else {
-        info!("Successfully synced all data to S3.");
+    // Sync object store
+    let mut object_store_joinset = JoinSet::new();
+    sync_all_streams(&mut object_store_joinset);
+
+    while let Some(res) = object_store_joinset.join_next().await {
+        match res {
+            Ok(Ok(_)) => info!("Successfully synced all data to S3."),
+            Ok(Err(err)) => error!("Failed to sync local data with object store. {err:?}"),
+            Err(err) => error!("Failed to join async task: {err}"),
+        }
     }
 }
 

src/handlers/http/modal/query_server.rs

@@ -27,6 +27,7 @@ use crate::handlers::http::{logstream, MAX_EVENT_PAYLOAD_SIZE};
 use crate::handlers::http::{rbac, role};
 use crate::hottier::HotTierManager;
 use crate::rbac::role::Action;
+use crate::sync::sync_start;
 use crate::{analytics, migration, storage, sync};
 use actix_web::web::{resource, ServiceConfig};
 use actix_web::{web, Scope};
@@ -126,6 +127,13 @@ impl ParseableServer for QueryServer {
         if init_cluster_metrics_schedular().is_ok() {
             info!("Cluster metrics scheduler started successfully");
         }
+
+        // local sync on init
+        tokio::spawn(async {
+            if let Err(e) = sync_start().await {
+                tracing::warn!("local sync on server start failed: {e}");
+            }
+        });
         if let Some(hot_tier_manager) = HotTierManager::global() {
             hot_tier_manager.put_internal_stream_hot_tier().await?;
             hot_tier_manager.download_from_s3()?;

src/handlers/http/modal/server.rs

@@ -33,6 +33,7 @@ use crate::metrics;
 use crate::migration;
 use crate::storage;
 use crate::sync;
+use crate::sync::sync_start;
 
 use actix_web::web;
 use actix_web::web::resource;
@@ -122,6 +123,13 @@ impl ParseableServer for Server {
 
         storage::retention::load_retention_from_global();
 
+        // local sync on init
+        tokio::spawn(async {
+            if let Err(e) = sync_start().await {
+                tracing::warn!("local sync on server start failed: {e}");
+            }
+        });
+
         if let Some(hot_tier_manager) = HotTierManager::global() {
             hot_tier_manager.download_from_s3()?;
         };

src/parseable/staging/reader.rs

@@ -30,7 +30,7 @@ use arrow_ipc::{reader::StreamReader, root_as_message_unchecked, MessageHeader};
 use arrow_schema::Schema;
 use byteorder::{LittleEndian, ReadBytesExt};
 use itertools::kmerge_by;
-use tracing::{error, warn};
+use tracing::error;
 
 use crate::{
     event::DEFAULT_TIMESTAMP_KEY,
@@ -85,20 +85,22 @@ impl MergedReverseRecordReader {
     pub fn try_new(file_paths: &[PathBuf]) -> Self {
         let mut readers = Vec::with_capacity(file_paths.len());
         for path in file_paths {
-            let Ok(file) = File::open(path) else {
-                warn!("Error when trying to read file: {path:?}");
-                continue;
-            };
-
-            let reader = match get_reverse_reader(file) {
-                Ok(r) => r,
+            match File::open(path) {
                 Err(err) => {
-                    error!("Invalid file detected, ignoring it: {path:?}; error = {err}");
+                    error!("Error when trying to read file: {path:?}; error = {err}");
                     continue;
                 }
-            };
-
-            readers.push(reader);
+                Ok(file) => {
+                    let reader = match get_reverse_reader(file) {
+                        Ok(r) => r,
+                        Err(err) => {
+                            error!("Invalid file detected, ignoring it: {path:?}; error = {err}");
+                            continue;
+                        }
+                    };
+                    readers.push(reader);
+                }
+            }
         }
 
         Self { readers }