fix: ensure panic safety

src/parseable/mod.rs

@@ -58,6 +58,9 @@ use crate::{
 mod staging;
 mod streams;
 
+/// File extension for arrow files in staging
+const ARROW_FILE_EXTENSION: &str = "arrows";
+
 /// Name of a Stream
 /// NOTE: this used to be a struct, flattened out for simplicity
 pub type LogStream = String;

src/parseable/streams.rs

@@ -61,15 +61,13 @@ use super::{
         writer::Writer,
         StagingError,
     },
-    LogStream,
+    LogStream, ARROW_FILE_EXTENSION,
 };
 
 #[derive(Debug, thiserror::Error)]
 #[error("Stream not found: {0}")]
 pub struct StreamNotFound(pub String);
 
-const ARROW_FILE_EXTENSION: &str = "data.arrows";
-
 pub type StreamRef = Arc<Stream>;
 
 /// All state associated with a single logstream in Parseable.
@@ -156,7 +154,7 @@ impl Stream {
             hostname.push_str(id);
         }
         let filename = format!(
-            "{}{stream_hash}.date={}.hour={:02}.minute={}.{}{hostname}.{ARROW_FILE_EXTENSION}",
+            "{}{stream_hash}.date={}.hour={:02}.minute={}.{}{hostname}.data.{ARROW_FILE_EXTENSION}",
             Utc::now().format("%Y%m%dT%H%M"),
             parsed_timestamp.date(),
             parsed_timestamp.hour(),
@@ -486,10 +484,12 @@ impl Stream {
             }
             writer.close()?;
 
-            if part_file.metadata().unwrap().len() < parquet::file::FOOTER_SIZE as u64 {
+            if part_file.metadata().expect("File was just created").len()
+                < parquet::file::FOOTER_SIZE as u64
+            {
                 error!(
-                    "Invalid parquet file {:?} detected for stream {}, removing it",
-                    &part_path, &self.stream_name
+                    "Invalid parquet file {part_path:?} detected for stream {}, removing it",
+                    &self.stream_name
                 );
                 remove_file(part_path).unwrap();
             } else {
@@ -501,15 +501,22 @@ impl Stream {
                 }
 
                 for file in arrow_files {
-                    // warn!("file-\n{file:?}\n");
-                    let file_size = file.metadata().unwrap().len();
-                    let file_type = file.extension().unwrap().to_str().unwrap();
-                    if remove_file(file.clone()).is_err() {
+                    let file_size = match file.metadata() {
+                        Ok(meta) => meta.len(),
+                        Err(err) => {
+                            warn!(
+                                "File ({}) not found; Error = {err}",
+                                file.display()
+                            );
+                            continue;
+                        }
+                    };
+                    if remove_file(&file).is_err() {
                         error!("Failed to delete file. Unstable state");
                         process::abort()
                     }
                     metrics::STORAGE_SIZE
-                        .with_label_values(&["staging", &self.stream_name, file_type])
+                        .with_label_values(&["staging", &self.stream_name, ARROW_FILE_EXTENSION])
                         .sub(file_size as i64);
                 }
             }
@@ -874,7 +881,7 @@ mod tests {
         );
 
         let expected_path = staging.data_path.join(format!(
-            "{}{stream_hash}.date={}.hour={:02}.minute={}.{}.{ARROW_FILE_EXTENSION}",
+            "{}{stream_hash}.date={}.hour={:02}.minute={}.{}.data.{ARROW_FILE_EXTENSION}",
             Utc::now().format("%Y%m%dT%H%M"),
             parsed_timestamp.date(),
             parsed_timestamp.hour(),
@@ -909,7 +916,7 @@ mod tests {
         );
 
         let expected_path = staging.data_path.join(format!(
-            "{}{stream_hash}.date={}.hour={:02}.minute={}.key1=value1.key2=value2.{}.{ARROW_FILE_EXTENSION}",
+            "{}{stream_hash}.date={}.hour={:02}.minute={}.key1=value1.key2=value2.{}.data.{ARROW_FILE_EXTENSION}",
             Utc::now().format("%Y%m%dT%H%M"),
             parsed_timestamp.date(),
             parsed_timestamp.hour(),