coderabbit suggestions

Author: parmesant

src/handlers/http/logstream.rs

@@ -473,6 +473,19 @@ pub async fn delete_stream_hot_tier(
 
     hot_tier_manager.delete_hot_tier(&stream_name).await?;
 
+    let mut stream_metadata: ObjectStoreFormat = serde_json::from_slice(
+        &PARSEABLE
+            .metastore
+            .get_stream_json(&stream_name, false)
+            .await?,
+    )?;
+    stream_metadata.hot_tier_enabled = false;
+
+    PARSEABLE
+        .metastore
+        .put_stream_json(&stream_metadata, &stream_name)
+        .await?;
+
     Ok((
         format!("hot tier deleted for stream {stream_name}"),
         StatusCode::OK,

src/storage/localfs.rs

@@ -139,18 +139,29 @@ impl ObjectStorage for LocalFS {
     }
     async fn get_object(&self, path: &RelativePath) -> Result<Bytes, ObjectStorageError> {
         let time = Instant::now();
-        let file_path = if path.to_string().contains(&self.root.to_str().unwrap()[1..]) {
-            #[cfg(windows)]
-            {
-                path.to_path("")
-            }
-            #[cfg(not(windows))]
-            {
+
+        let file_path;
+
+        // this is for the `get_manifest()` function because inside a snapshot, we store the absolute path (without `/`) on linux based OS
+        // `home/user/.../manifest.json`
+        // on windows, the path is stored with the drive letter
+        // `D:\\parseable\\data..\\manifest.json`
+        // thus, we need to check if the root of localfs is already present in the path
+        #[cfg(windows)]
+        {
+            // in windows the absolute path (self.root) doesn't matter because we store the complete path
+            file_path = path.to_path("");
+        }
+        #[cfg(not(windows))]
+        {
+            // absolute path (self.root) will always start with `/`
+            let root_str = self.root.to_str().unwrap();
+            file_path = if path.to_string().contains(&root_str[1..]) && root_str.len() > 1 {
                 path.to_path("/")
-            }
-        } else {
-            self.path_in_root(path)
-        };
+            } else {
+                self.path_in_root(path)
+            };
+        }
 
         let res: Result<Bytes, ObjectStorageError> = match fs::read(file_path).await {
             Ok(x) => Ok(x.into()),

src/storage/object_storage.rs

@@ -280,7 +280,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
         meta: ObjectStoreFormat,
         schema: Arc<Schema>,
     ) -> Result<String, ObjectStorageError> {
-        let s = &*schema.clone();
+        let s: Schema = schema.as_ref().clone();
         PARSEABLE
             .metastore
             .put_schema(s.clone(), stream_name)
@@ -955,7 +955,7 @@ fn stream_relative_path(
 }
 
 pub fn sync_all_streams(joinset: &mut JoinSet<Result<(), ObjectStorageError>>) {
-    let object_store = PARSEABLE.storage.get_object_store();
+    let object_store = PARSEABLE.storage().get_object_store();
     for stream_name in PARSEABLE.streams.list() {
         let object_store = object_store.clone();
         joinset.spawn(async move {
@@ -989,7 +989,7 @@ pub async fn commit_schema_to_storage(
         schema,
         serde_json::from_slice::<Schema>(&stream_schema)?,
     ])
-    .unwrap();
+    .map_err(|e| ObjectStorageError::Custom(e.to_string()))?;
 
     PARSEABLE
         .metastore