Proper Error Message if the Storage is in an invalid state

server/src/main.rs

@@ -58,6 +58,7 @@ async fn main() -> anyhow::Result<()> {
     migration::run_metadata_migration(&CONFIG).await?;
     let metadata = storage::resolve_parseable_metadata().await?;
     CONFIG.validate_staging()?;
+    CONFIG.validate_storage().await?;
     banner::print(&CONFIG, &metadata).await;
     rbac::map::init(&metadata);
     metadata.set_global();

server/src/option.rs

@@ -27,7 +27,7 @@ use std::sync::Arc;
 use url::Url;
 
 use crate::oidc::{self, OpenidConfig};
-use crate::storage::{FSConfig, ObjectStorageProvider, S3Config};
+use crate::storage::{FSConfig, ObjectStorageError, ObjectStorageProvider, S3Config};
 use crate::utils::validate_path_is_writeable;
 
 pub const MIN_CACHE_SIZE_BYTES: u64 = 1000u64.pow(3); // 1 GiB
@@ -104,6 +104,37 @@ impl Config {
         validate_path_is_writeable(staging_path)
     }
 
+    pub async fn validate_storage(&self) -> Result<(), ObjectStorageError> {
+        let obj_store = self.storage.get_object_store();
+        let rel_path = relative_path::RelativePathBuf::from(".parseable.json");
+
+        let has_parseable_json = obj_store.get_object(&rel_path).await.is_ok();
+
+        let has_dirs = match obj_store.list_dirs_in_storage().await {
+            Ok(dirs) => dirs.is_empty(),
+            Err(_) => false,
+        };
+
+        let has_streams = obj_store.list_streams().await.is_ok();
+
+        if has_dirs || has_parseable_json && has_streams {
+            Ok(())
+        } else if has_parseable_json && !has_streams {
+            Err(ObjectStorageError::Custom(
+                "Parseable config found, but the Storage contains some stale data.".to_owned(),
+            ))
+        } else if !has_parseable_json && !has_streams && !has_dirs {
+            Err(ObjectStorageError::Custom(
+                "Storage contains some stale data, Please provide an Empty Storage".to_owned(),
+            ))
+        } else {
+            Err(ObjectStorageError::Custom(
+                "Parseable config is missing, but streams are present in Storage.\nJoin us on Parseable Slack"
+                .to_owned()
+            ))
+        }
+    }
+
     pub fn storage(&self) -> Arc<dyn ObjectStorageProvider + Send + Sync> {
         self.storage.clone()
     }

server/src/storage/localfs.rs

@@ -169,6 +169,23 @@ impl ObjectStorage for LocalFS {
         Ok(logstreams)
     }
 
+    async fn list_dirs_in_storage(&self) -> Result<Vec<String>, ObjectStorageError> {
+        let dirs = ReadDirStream::new(fs::read_dir(&self.root).await?)
+            .try_collect::<Vec<DirEntry>>()
+            .await?
+            .into_iter()
+            .map(dir_name);
+
+        let dirs = FuturesUnordered::from_iter(dirs)
+            .try_collect::<Vec<_>>()
+            .await?
+            .into_iter()
+            .flatten()
+            .collect::<Vec<_>>();
+
+        Ok(dirs)
+    }
+
     async fn list_dates(&self, stream_name: &str) -> Result<Vec<String>, ObjectStorageError> {
         let path = self.root.join(stream_name);
         let directories = ReadDirStream::new(fs::read_dir(&path).await?);

server/src/storage/object_storage.rs

@@ -75,6 +75,7 @@ pub trait ObjectStorage: Sync + 'static {
     async fn check(&self) -> Result<(), ObjectStorageError>;
     async fn delete_stream(&self, stream_name: &str) -> Result<(), ObjectStorageError>;
     async fn list_streams(&self) -> Result<Vec<LogStream>, ObjectStorageError>;
+    async fn list_dirs_in_storage(&self) -> Result<Vec<String>, ObjectStorageError>;
     async fn list_dates(&self, stream_name: &str) -> Result<Vec<String>, ObjectStorageError>;
     async fn upload_file(&self, key: &str, path: &Path) -> Result<(), ObjectStorageError>;
 

server/src/storage/s3.rs

@@ -470,6 +470,18 @@ impl ObjectStorage for S3 {
     fn store_url(&self) -> url::Url {
         url::Url::parse(&format!("s3://{}", self.bucket)).unwrap()
     }
+
+    async fn list_dirs_in_storage(&self) -> Result<Vec<String>, ObjectStorageError> {
+        let pre = object_store::path::Path::from("/");
+        let resp = self.client.list_with_delimiter(Some(&pre)).await?;
+
+        Ok(resp
+            .common_prefixes
+            .iter()
+            .flat_map(|path| path.parts())
+            .map(|name| name.as_ref().to_string())
+            .collect::<Vec<_>>())
+    }
 }
 
 impl From<object_store::Error> for ObjectStorageError {