Coderabbit suggestions

parmesant · parmesant · commit 2b613a48e62e · 2025-09-06T12:26:56.000+05:30
diff --git a/src/alerts/mod.rs b/src/alerts/mod.rs
@@ -995,7 +995,7 @@ impl AlertManagerTrait for Alerts {
         let mut map = self.alerts.write().await;
 
         // Get alerts path and read raw bytes for migration handling
-        let raw_objects = PARSEABLE.metastore.get_alerts().await.unwrap_or_default();
+        let raw_objects = PARSEABLE.metastore.get_alerts().await?;
 
         for raw_bytes in raw_objects {
             // First, try to parse as JSON Value to check version
diff --git a/src/alerts/target.rs b/src/alerts/target.rs
@@ -58,17 +58,17 @@ impl TargetConfigs {
     pub async fn load(&self) -> anyhow::Result<()> {
         let mut map = self.target_configs.write().await;
 
-        for target in PARSEABLE.metastore.get_targets().await.unwrap_or_default() {
+        for target in PARSEABLE.metastore.get_targets().await? {
             map.insert(target.id, target);
         }
 
         Ok(())
     }
 
     pub async fn update(&self, target: Target) -> Result<(), AlertError> {
+        PARSEABLE.metastore.put_target(&target).await?;
         let mut map = self.target_configs.write().await;
         map.insert(target.id, target.clone());
-        PARSEABLE.metastore.put_target(&target).await?;
         Ok(())
     }
 
diff --git a/src/handlers/http/alerts.rs b/src/handlers/http/alerts.rs
@@ -208,15 +208,15 @@ pub async fn post(
 
     alert.validate(&session_key).await?;
 
-    // now that we've validated that the user can run this query
-    // move on to saving the alert in ObjectStore
-    alerts.update(alert).await;
-
+    // update persistent storage first
     PARSEABLE
         .metastore
         .put_alert(&alert.to_alert_config())
         .await?;
 
+    // update in memory
+    alerts.update(alert).await;
+
     // start the task
     alerts.start_task(alert.clone_box()).await?;
 
diff --git a/src/handlers/http/modal/utils/rbac_utils.rs b/src/handlers/http/modal/utils/rbac_utils.rs
@@ -27,7 +27,7 @@ pub async fn get_metadata() -> Result<crate::storage::StorageMetadata, ObjectSto
         .get_parseable_metadata()
         .await
         .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?
-        .expect("metadata is initialized");
+        .ok_or_else(|| ObjectStorageError::Custom("parseable metadata not initialized".into()))?;
     Ok(serde_json::from_slice::<StorageMetadata>(&metadata)?)
 }
 
diff --git a/src/handlers/http/oidc.rs b/src/handlers/http/oidc.rs
@@ -448,7 +448,7 @@ async fn get_metadata() -> Result<crate::storage::StorageMetadata, ObjectStorage
         .get_parseable_metadata()
         .await
         .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?
-        .expect("metadata is initialized");
+        .ok_or_else(|| ObjectStorageError::Custom("parseable metadata not initialized".into()))?;
     Ok(serde_json::from_slice::<StorageMetadata>(&metadata)?)
 }
 
diff --git a/src/handlers/http/role.rs b/src/handlers/http/role.rs
@@ -146,7 +146,7 @@ async fn get_metadata() -> Result<crate::storage::StorageMetadata, ObjectStorage
         .get_parseable_metadata()
         .await
         .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?
-        .expect("metadata is initialized");
+        .ok_or_else(|| ObjectStorageError::Custom("parseable metadata not initialized".into()))?;
     Ok(serde_json::from_slice::<StorageMetadata>(&metadata)?)
 }
 
diff --git a/src/metastore/metastore_traits.rs b/src/metastore/metastore_traits.rs
@@ -36,8 +36,6 @@ use crate::{
 #[async_trait]
 pub trait Metastore: std::fmt::Debug + Send + Sync {
     async fn initiate_connection(&self) -> Result<(), MetastoreError>;
-    async fn list_objects(&self) -> Result<(), MetastoreError>;
-    async fn get_object(&self) -> Result<(), MetastoreError>;
     async fn get_objects(&self, parent_path: &str) -> Result<Vec<Bytes>, MetastoreError>;
 
     /// alerts
diff --git a/src/metastore/metastores/object_store_metastore.rs b/src/metastore/metastores/object_store_metastore.rs
@@ -68,16 +68,6 @@ impl Metastore for ObjectStoreMetastore {
         unimplemented!()
     }
 
-    /// Might implement later
-    async fn list_objects(&self) -> Result<(), MetastoreError> {
-        unimplemented!()
-    }
-
-    /// Might implement later
-    async fn get_object(&self) -> Result<(), MetastoreError> {
-        unimplemented!()
-    }
-
     /// Fetch mutiple .json objects
     async fn get_objects(&self, parent_path: &str) -> Result<Vec<Bytes>, MetastoreError> {
         Ok(self
@@ -105,7 +95,12 @@ impl Metastore for ObjectStoreMetastore {
 
     /// This function puts an alert in the object store at the given path
     async fn put_alert(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError> {
-        let path = alert_json_path(Ulid::from_string(&obj.get_object_id()).unwrap());
+        let id = Ulid::from_string(&obj.get_object_id()).map_err(|e| MetastoreError::Error {
+            status_code: StatusCode::BAD_REQUEST,
+            message: e.to_string(),
+            flow: "put_alert".into(),
+        })?;
+        let path = alert_json_path(id);
 
         Ok(self.storage.put_object(&path, to_bytes(obj)).await?)
     }
@@ -536,7 +531,10 @@ impl Metastore for ObjectStoreMetastore {
             .await?
             .iter()
             // we should be able to unwrap as we know the data is valid schema
-            .map(|byte_obj| serde_json::from_slice(byte_obj).expect("data is valid json"))
+            .map(|byte_obj| {
+                serde_json::from_slice(byte_obj)
+                    .unwrap_or_else(|_| panic!("got an invalid schema for stream: {stream_name}"))
+            })
             .collect())
     }
 
@@ -661,6 +659,7 @@ impl Metastore for ObjectStoreMetastore {
                 .await
                 .map_err(MetastoreError::ObjectStorageError)
         } else {
+            // not local-disk, object storage
             let mut result_file_list = HashSet::new();
             let resp = self.storage.list_with_delimiter(None).await?;
 
@@ -669,7 +668,12 @@ impl Metastore for ObjectStoreMetastore {
                 .iter()
                 .flat_map(|path| path.parts())
                 .map(|name| name.as_ref().to_string())
-                .filter(|name| name != PARSEABLE_ROOT_DIRECTORY && name != USERS_ROOT_DIR)
+                .filter(|name| {
+                    name != PARSEABLE_ROOT_DIRECTORY
+                        && name != USERS_ROOT_DIR
+                        && name != SETTINGS_ROOT_DIRECTORY
+                        && name != ALERTS_ROOT_DIRECTORY
+                })
                 .collect::<Vec<_>>();
 
             for stream in streams {
diff --git a/src/metastore/mod.rs b/src/metastore/mod.rs
@@ -95,7 +95,7 @@ impl MetastoreError {
                 file_path: None,
                 timestamp: Some(chrono::Utc::now()),
                 metadata: std::collections::HashMap::new(),
-                status_code: 500,
+                status_code: 400,
             },
             MetastoreError::JsonSchemaError { message } => MetastoreErrorDetail {
                 operation: "JsonSchemaError".to_string(),
@@ -104,7 +104,7 @@ impl MetastoreError {
                 file_path: None,
                 timestamp: Some(chrono::Utc::now()),
                 metadata: std::collections::HashMap::new(),
-                status_code: 500,
+                status_code: 400,
             },
             MetastoreError::InvalidJsonStructure { expected, found } => MetastoreErrorDetail {
                 operation: "InvalidJsonStructure".to_string(),
@@ -118,7 +118,7 @@ impl MetastoreError {
                 ]
                 .into_iter()
                 .collect(),
-                status_code: 500,
+                status_code: 400,
             },
             MetastoreError::MissingJsonField { field } => MetastoreErrorDetail {
                 operation: "MissingJsonField".to_string(),
@@ -127,7 +127,7 @@ impl MetastoreError {
                 file_path: None,
                 timestamp: Some(chrono::Utc::now()),
                 metadata: [("field".to_string(), field.clone())].into_iter().collect(),
-                status_code: 500,
+                status_code: 400,
             },
             MetastoreError::InvalidJsonValue { field, reason } => MetastoreErrorDetail {
                 operation: "InvalidJsonValue".to_string(),
@@ -141,7 +141,7 @@ impl MetastoreError {
                 ]
                 .into_iter()
                 .collect(),
-                status_code: 500,
+                status_code: 400,
             },
         }
     }
diff --git a/src/parseable/streams.rs b/src/parseable/streams.rs
@@ -660,7 +660,7 @@ impl Stream {
             return Ok(None);
         }
 
-        Ok(Some(Schema::try_merge(schemas).unwrap()))
+        Ok(Some(Schema::try_merge(schemas)?))
     }
 
     fn write_parquet_part_file(
diff --git a/src/query/mod.rs b/src/query/mod.rs
@@ -43,7 +43,6 @@ use std::ops::Bound;
 use std::sync::Arc;
 use sysinfo::System;
 use tokio::runtime::Runtime;
-use tracing::warn;
 
 use self::error::ExecuteError;
 use self::stream_schema_provider::GlobalSchemaProvider;
@@ -547,38 +546,26 @@ pub async fn get_manifest_list(
         PartialTimeFilter::High(Bound::Included(time_range.end.naive_utc())),
     ];
 
-    warn!(merged_snapshot=?merged_snapshot);
-    warn!(time_filter=?time_filter);
-
     let mut all_manifest_files = Vec::new();
     for manifest_item in merged_snapshot.manifests(&time_filter) {
-        warn!(manifest_item=?manifest_item);
-        all_manifest_files.push(
-            PARSEABLE
-                .metastore
-                .get_manifest(
-                    stream_name,
-                    manifest_item.time_lower_bound,
-                    manifest_item.time_upper_bound,
-                    Some(manifest_item.manifest_path),
-                )
-                .await?
-                .expect("Data is invalid for Manifest"),
-        );
+        let manifest_opt = PARSEABLE
+            .metastore
+            .get_manifest(
+                stream_name,
+                manifest_item.time_lower_bound,
+                manifest_item.time_upper_bound,
+                Some(manifest_item.manifest_path),
+            )
+            .await?;
+        let manifest = manifest_opt.ok_or_else(|| {
+            QueryError::CustomError(format!(
+                "Manifest not found for {stream_name} [{} - {}]",
+                manifest_item.time_lower_bound, manifest_item.time_upper_bound
+            ))
+        })?;
+        all_manifest_files.push(manifest);
     }
 
-    // let all_manifest_files = collect_manifest_files(
-    //     object_store,
-    //     merged_snapshot
-    //         .manifests(&time_filter)
-    //         .into_iter()
-    //         .sorted_by_key(|file| file.time_lower_bound)
-    //         .map(|item| item.manifest_path)
-    //         .collect(),
-    // )
-    // .await
-    // .map_err(|err| anyhow::Error::msg(err.to_string()))?;
-
     Ok(all_manifest_files)
 }
 
diff --git a/src/query/stream_schema_provider.rs b/src/query/stream_schema_provider.rs
@@ -414,19 +414,26 @@ async fn collect_from_snapshot(
     let mut manifest_files = Vec::new();
 
     for manifest_item in snapshot.manifests(time_filters) {
-        manifest_files.push(
-            PARSEABLE
-                .metastore
-                .get_manifest(
-                    stream_name,
-                    manifest_item.time_lower_bound,
-                    manifest_item.time_upper_bound,
-                    Some(manifest_item.manifest_path),
-                )
-                .await
-                .map_err(|e| DataFusionError::Plan(e.to_string()))?
-                .expect("Data is invalid for Manifest"),
-        )
+        let manifest_opt = PARSEABLE
+            .metastore
+            .get_manifest(
+                stream_name,
+                manifest_item.time_lower_bound,
+                manifest_item.time_upper_bound,
+                Some(manifest_item.manifest_path),
+            )
+            .await
+            .map_err(|e| DataFusionError::Plan(e.to_string()))?;
+        if let Some(manifest) = manifest_opt {
+            manifest_files.push(manifest);
+        } else {
+            tracing::warn!(
+                "Manifest missing for stream={} [{:?} - {:?}]",
+                stream_name,
+                manifest_item.time_lower_bound,
+                manifest_item.time_upper_bound
+            );
+        }
     }
 
     let mut manifest_files: Vec<_> = manifest_files
diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs
@@ -274,17 +274,6 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
         prefix: Option<object_store::path::Path>,
     ) -> Result<ListResult, ObjectStorageError>;
 
-    // async fn put_schema(
-    //     &self,
-    //     stream_name: &str,
-    //     schema: &Schema,
-    // ) -> Result<(), ObjectStorageError> {
-    //     self.put_object(&schema_path(stream_name), to_bytes(schema))
-    //         .await?;
-
-    //     Ok(())
-    // }
-
     async fn create_stream(
         &self,
         stream_name: &str,

Original file line number	Diff line number	Diff line change
`@@ -58,17 +58,17 @@ impl TargetConfigs {`
`58`	`58`	`pub async fn load(&self) -> anyhow::Result<()> {`
`59`	`59`	`let mut map = self.target_configs.write().await;`
`60`	`60`
`61`		`- for target in PARSEABLE.metastore.get_targets().await.unwrap_or_default() {`
	`61`	`+ for target in PARSEABLE.metastore.get_targets().await? {`
`62`	`62`	`map.insert(target.id, target);`
`63`	`63`	`}`
`64`	`64`
`65`	`65`	`Ok(())`
`66`	`66`	`}`
`67`	`67`
`68`	`68`	`pub async fn update(&self, target: Target) -> Result<(), AlertError> {`
	`69`	`+ PARSEABLE.metastore.put_target(&target).await?;`
`69`	`70`	`let mut map = self.target_configs.write().await;`
`70`	`71`	`map.insert(target.id, target.clone());`
`71`		`- PARSEABLE.metastore.put_target(&target).await?;`
`72`	`72`	`Ok(())`
`73`	`73`	`}`
`74`	`74`
Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ pub async fn get_metadata() -> Result<crate::storage::StorageMetadata, ObjectSto`
`27`	`27`	`.get_parseable_metadata()`
`28`	`28`	`.await`
`29`	`29`	`.map_err(\|e\| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?`
`30`		`- .expect("metadata is initialized");`
	`30`	`+ .ok_or_else(\|\| ObjectStorageError::Custom("parseable metadata not initialized".into()))?;`
`31`	`31`	`Ok(serde_json::from_slice::<StorageMetadata>(&metadata)?)`
`32`	`32`	`}`
`33`	`33`
Original file line number	Diff line number	Diff line change
`@@ -448,7 +448,7 @@ async fn get_metadata() -> Result<crate::storage::StorageMetadata, ObjectStorage`
`448`	`448`	`.get_parseable_metadata()`
`449`	`449`	`.await`
`450`	`450`	`.map_err(\|e\| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?`
`451`		`- .expect("metadata is initialized");`
	`451`	`+ .ok_or_else(\|\| ObjectStorageError::Custom("parseable metadata not initialized".into()))?;`
`452`	`452`	`Ok(serde_json::from_slice::<StorageMetadata>(&metadata)?)`
`453`	`453`	`}`
`454`	`454`
Original file line number	Diff line number	Diff line change
`@@ -146,7 +146,7 @@ async fn get_metadata() -> Result<crate::storage::StorageMetadata, ObjectStorage`
`146`	`146`	`.get_parseable_metadata()`
`147`	`147`	`.await`
`148`	`148`	`.map_err(\|e\| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?`
`149`		`- .expect("metadata is initialized");`
	`149`	`+ .ok_or_else(\|\| ObjectStorageError::Custom("parseable metadata not initialized".into()))?;`
`150`	`150`	`Ok(serde_json::from_slice::<StorageMetadata>(&metadata)?)`
`151`	`151`	`}`
`152`	`152`
Original file line number	Diff line number	Diff line change
`@@ -660,7 +660,7 @@ impl Stream {`
`660`	`660`	`return Ok(None);`
`661`	`661`	`}`
`662`	`662`
`663`		`- Ok(Some(Schema::try_merge(schemas).unwrap()))`
	`663`	`+ Ok(Some(Schema::try_merge(schemas)?))`
`664`	`664`	`}`
`665`	`665`
`666`	`666`	`fn write_parquet_part_file(`