fix: Multiple fixes

- remove invalid tenant dir
- log sync errors
- add tenant to filter

Author: parmesant

src/handlers/http/modal/query/querier_logstream.rs

@@ -157,7 +157,9 @@ pub async fn put_stream(
             actix_web::http::header::HeaderValue::from_str(&userid).unwrap(),
         );
     }
-    sync_streams_with_ingestors(headers, body, &stream_name, &tenant_id).await?;
+    if let Err(e) = sync_streams_with_ingestors(headers, body, &stream_name, &tenant_id).await {
+        tracing::error!("{e}");
+    };
 
     if is_update {
         Ok(("Log stream updated", StatusCode::OK))

src/handlers/http/modal/query/querier_rbac.rs

@@ -92,7 +92,9 @@ pub async fn post_user(
     // let created_role = user_roles.clone();
     Users.put_user(user.clone());
 
-    sync_user_creation(&req, user, &None, &tenant_id, &caller_userid).await?;
+    if let Err(e) = sync_user_creation(&req, user, &None, &tenant_id, &caller_userid).await {
+        tracing::error!("{e}");
+    };
 
     Ok(password)
 }
@@ -161,7 +163,11 @@ pub async fn delete_user(
     }
     put_metadata(&metadata, &tenant_id).await?;
 
-    sync_user_deletion_with_ingestors(&req, &userid, &tenant_id, &caller_userid).await?;
+    if let Err(e) =
+        sync_user_deletion_with_ingestors(&req, &userid, &tenant_id, &caller_userid).await
+    {
+        tracing::error!("{e}");
+    };
 
     // update in mem table
     Users.delete_user(&userid, &tenant_id);
@@ -380,7 +386,9 @@ pub async fn post_gen_password(
     put_metadata(&metadata, &tenant_id).await?;
     Users.change_password_hash(&username, &new_hash, &tenant_id);
 
-    sync_password_reset_with_ingestors(req, &username, &caller_userid).await?;
+    if let Err(e) = sync_password_reset_with_ingestors(req, &username, &caller_userid).await {
+        tracing::error!("{e}");
+    };
 
     Ok(new_password)
 }

src/handlers/http/users/filters.rs

@@ -71,6 +71,7 @@ pub async fn post(
     filter.filter_id = Some(filter_id.clone());
     filter.user_id = Some(user_id.clone());
     filter.version = Some(CURRENT_FILTER_VERSION.to_string());
+    filter.tenant_id = tenant_id.clone();
     PARSEABLE.metastore.put_filter(&filter, &tenant_id).await?;
     FILTERS.update(&filter, &tenant_id).await;
 

src/parseable/mod.rs

@@ -1161,9 +1161,20 @@ impl Parseable {
                 TENANT_METADATA.insert_tenant(tenant_id.clone(), metadata.clone());
             } else if !is_multi_tenant {
             } else {
-                return Err(anyhow::Error::msg(format!(
-                    "Found invalid tenant directory with multi-tenant mode- {tenant_id}.\nExiting."
-                )));
+                // found an invalid multi-tenant directory, delete
+                tracing::warn!(
+                    "Found invalid tenant directory with multi-tenant mode- {tenant_id}.\nDeleting."
+                );
+
+                // delete from storage
+                if let Err(e) = obj_store
+                    .delete_prefix(&RelativePathBuf::from_iter([tenant_id]), &None)
+                    .await
+                {
+                    tracing::error!(
+                        "Unable to delete invalid directory- {tenant_id} with error-\n{e}"
+                    );
+                };
             }
         }