fix: centralize tenant-aware object path construction

src/catalog/mod.rs

@@ -40,7 +40,7 @@ use crate::{
     stats::{event_labels_date, get_current_stats, storage_size_labels_date, update_deleted_stats},
     storage::{
         ObjectStorage, ObjectStorageError, ObjectStoreFormat,
-        object_storage::{manifest_segment_matches, own_manifest_file_name},
+        object_storage::{manifest_segment_matches, own_manifest_file_name, tenant_prefix},
     },
 };
 pub use manifest::create_from_parquet_file;
@@ -568,7 +568,7 @@ pub fn partition_path(
     upper_bound: DateTime<Utc>,
     tenant_id: &Option<String>,
 ) -> RelativePathBuf {
-    let root = tenant_id.as_deref().unwrap_or("");
+    let root = tenant_prefix(tenant_id);
     let lower = lower_bound.date_naive().format("%Y-%m-%d").to_string();
     let upper = upper_bound.date_naive().format("%Y-%m-%d").to_string();
     if lower == upper {

src/metastore/metastores/object_store_metastore.rs

@@ -59,7 +59,7 @@ use crate::{
         object_storage::{
             alert_json_path, alert_state_json_path, filter_path, manifest_path, mttr_json_path,
             outbound_http_policy_json_path, parseable_json_path, schema_path, stream_json_path,
-            to_bytes,
+            tenant_prefix, to_bytes,
         },
     },
     users::filters::{Filter, migrate_v1_v2},
@@ -777,7 +777,7 @@ impl Metastore for ObjectStoreMetastore {
         tenant_id: &Option<String>,
         _is_migration: bool,
     ) -> Result<Bytes, MetastoreError> {
-        let tenant = tenant_id.as_deref().unwrap_or("");
+        let tenant = tenant_prefix(tenant_id);
         let path = if get_base {
             RelativePathBuf::from_iter([
                 tenant,
@@ -799,7 +799,7 @@ impl Metastore for ObjectStoreMetastore {
         tenant_id: &Option<String>,
         _is_migration: bool,
     ) -> Result<Vec<Bytes>, MetastoreError> {
-        let root = tenant_id.as_deref().unwrap_or("");
+        let root = tenant_prefix(tenant_id);
         let path = RelativePathBuf::from_iter([root, stream_name, STREAM_ROOT_DIRECTORY]);
         if let Some(mode) = mode {
             if mode.eq(&Mode::Ingest) {

src/storage/azure_blob.rs

@@ -63,7 +63,8 @@ use crate::{
 use super::{
     CONNECT_TIMEOUT_SECS, DEFAULT_MAX_OBJECT_STORE_REQUESTS, ObjectStorage, ObjectStorageError,
     ObjectStorageProvider, PARSEABLE_ROOT_DIRECTORY, REQUEST_TIMEOUT_SECS,
-    STREAM_METADATA_FILE_NAME, metrics_layer::MetricLayer, object_storage::parseable_json_path,
+    STREAM_METADATA_FILE_NAME, metrics_layer::MetricLayer,
+    object_storage::{parseable_json_path, stream_prefix},
     partial_path, to_object_store_path,
 };
 use crate::storage::GET_OBJECTS_CONCURRENCY;
@@ -459,9 +460,10 @@ impl BlobStore {
         stream: &str,
         tenant_id: &Option<String>,
     ) -> Result<Vec<String>, ObjectStorageError> {
+        let prefix = format!("{}/", stream_prefix(stream, tenant_id));
         let resp: Result<object_store::ListResult, object_store::Error> = self
             .client
-            .list_with_delimiter(Some(&(stream.into())))
+            .list_with_delimiter(Some(&object_store::path::Path::from(prefix.as_str())))
             .await;
         let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         increment_object_store_calls_by_date("LIST", &Utc::now().date_naive().to_string(), tenant);
@@ -485,7 +487,7 @@ impl BlobStore {
         // return prefixes at the root level
         let dates: Vec<_> = common_prefixes
             .iter()
-            .filter_map(|path| path.as_ref().strip_prefix(&format!("{stream}/")))
+            .filter_map(|path| path.as_ref().strip_prefix(&prefix))
             .map(String::from)
             .collect();
         tracing::Span::current().record("dates", dates.len());
@@ -889,12 +891,8 @@ impl ObjectStorage for BlobStore {
         stream_name: &str,
         tenant_id: &Option<String>,
     ) -> Result<(), ObjectStorageError> {
-        let prefix = if let Some(tenant) = tenant_id.as_ref() {
-            &format!("{tenant}/{stream_name}")
-        } else {
-            stream_name
-        };
-        self._delete_prefix(prefix, tenant_id).await?;
+        let prefix = stream_prefix(stream_name, tenant_id);
+        self._delete_prefix(&prefix, tenant_id).await?;
 
         Ok(())
     }
@@ -998,7 +996,8 @@ impl ObjectStorage for BlobStore {
         date: &str,
         tenant_id: &Option<String>,
     ) -> Result<Vec<String>, ObjectStorageError> {
-        let pre = object_store::path::Path::from(format!("{}/{}/", stream_name, date));
+        let prefix = format!("{}/{date}/", stream_prefix(stream_name, tenant_id));
+        let pre = object_store::path::Path::from(prefix.as_str());
         let resp = self.client.list_with_delimiter(Some(&pre)).await?;
         let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         increment_files_scanned_in_object_store_calls_by_date(
@@ -1014,8 +1013,7 @@ impl ObjectStorage for BlobStore {
             .iter()
             .filter_map(|path| {
                 let path_str = path.as_ref();
-                if let Some(stripped) = path_str.strip_prefix(&format!("{}/{}/", stream_name, date))
-                {
+                if let Some(stripped) = path_str.strip_prefix(&prefix) {
                     // Remove trailing slash if present, otherwise use as is
                     let clean_path = stripped.strip_suffix('/').unwrap_or(stripped);
                     Some(clean_path.to_string())
@@ -1036,7 +1034,8 @@ impl ObjectStorage for BlobStore {
         hour: &str,
         tenant_id: &Option<String>,
     ) -> Result<Vec<String>, ObjectStorageError> {
-        let pre = object_store::path::Path::from(format!("{}/{}/{}/", stream_name, date, hour));
+        let prefix = format!("{}/{date}/{hour}/", stream_prefix(stream_name, tenant_id));
+        let pre = object_store::path::Path::from(prefix.as_str());
         let resp = self.client.list_with_delimiter(Some(&pre)).await?;
         let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         increment_files_scanned_in_object_store_calls_by_date(
@@ -1051,9 +1050,7 @@ impl ObjectStorage for BlobStore {
             .iter()
             .filter_map(|path| {
                 let path_str = path.as_ref();
-                if let Some(stripped) =
-                    path_str.strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour))
-                {
+                if let Some(stripped) = path_str.strip_prefix(&prefix) {
                     // Remove trailing slash if present, otherwise use as is
                     let clean_path = stripped.strip_suffix('/').unwrap_or(stripped);
                     Some(clean_path.to_string())

src/storage/gcs.rs

@@ -62,7 +62,8 @@ use tracing::error;
 use super::{
     CONNECT_TIMEOUT_SECS, DEFAULT_MAX_OBJECT_STORE_REQUESTS, ObjectStorage, ObjectStorageError,
     ObjectStorageProvider, PARSEABLE_ROOT_DIRECTORY, REQUEST_TIMEOUT_SECS,
-    STREAM_METADATA_FILE_NAME, metrics_layer::MetricLayer, object_storage::parseable_json_path,
+    STREAM_METADATA_FILE_NAME, metrics_layer::MetricLayer,
+    object_storage::{parseable_json_path, stream_prefix},
     partial_path, to_object_store_path,
 };
 use crate::storage::GET_OBJECTS_CONCURRENCY;
@@ -425,9 +426,10 @@ impl Gcs {
         stream: &str,
         tenant_id: &Option<String>,
     ) -> Result<Vec<String>, ObjectStorageError> {
+        let prefix = format!("{}/", stream_prefix(stream, tenant_id));
         let resp: Result<object_store::ListResult, object_store::Error> = self
             .client
-            .list_with_delimiter(Some(&(stream.into())))
+            .list_with_delimiter(Some(&object_store::path::Path::from(prefix.as_str())))
             .await;
         let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         increment_object_store_calls_by_date("LIST", &Utc::now().date_naive().to_string(), tenant);
@@ -451,7 +453,7 @@ impl Gcs {
         // return prefixes at the root level
         let dates: Vec<_> = common_prefixes
             .iter()
-            .filter_map(|path| path.as_ref().strip_prefix(&format!("{stream}/")))
+            .filter_map(|path| path.as_ref().strip_prefix(&prefix))
             .map(String::from)
             .collect();
         tracing::Span::current().record("dates", dates.len());
@@ -872,12 +874,8 @@ impl ObjectStorage for Gcs {
         stream_name: &str,
         tenant_id: &Option<String>,
     ) -> Result<(), ObjectStorageError> {
-        let prefix = if let Some(tenant) = tenant_id.as_ref() {
-            &format!("{tenant}/{stream_name}")
-        } else {
-            stream_name
-        };
-        self._delete_prefix(prefix, tenant_id).await?;
+        let prefix = stream_prefix(stream_name, tenant_id);
+        self._delete_prefix(&prefix, tenant_id).await?;
 
         Ok(())
     }
@@ -980,7 +978,8 @@ impl ObjectStorage for Gcs {
         date: &str,
         tenant_id: &Option<String>,
     ) -> Result<Vec<String>, ObjectStorageError> {
-        let pre = object_store::path::Path::from(format!("{}/{}/", stream_name, date));
+        let prefix = format!("{}/{date}/", stream_prefix(stream_name, tenant_id));
+        let pre = object_store::path::Path::from(prefix.as_str());
         let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         let resp = self.client.list_with_delimiter(Some(&pre)).await?;
         increment_files_scanned_in_object_store_calls_by_date(
@@ -996,8 +995,7 @@ impl ObjectStorage for Gcs {
             .iter()
             .filter_map(|path| {
                 let path_str = path.as_ref();
-                if let Some(stripped) = path_str.strip_prefix(&format!("{}/{}/", stream_name, date))
-                {
+                if let Some(stripped) = path_str.strip_prefix(&prefix) {
                     // Remove trailing slash if present, otherwise use as is
                     let clean_path = stripped.strip_suffix('/').unwrap_or(stripped);
                     Some(clean_path.to_string())
@@ -1018,7 +1016,8 @@ impl ObjectStorage for Gcs {
         hour: &str,
         tenant_id: &Option<String>,
     ) -> Result<Vec<String>, ObjectStorageError> {
-        let pre = object_store::path::Path::from(format!("{}/{}/{}/", stream_name, date, hour));
+        let prefix = format!("{}/{date}/{hour}/", stream_prefix(stream_name, tenant_id));
+        let pre = object_store::path::Path::from(prefix.as_str());
         let resp = self.client.list_with_delimiter(Some(&pre)).await?;
         let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         increment_files_scanned_in_object_store_calls_by_date(
@@ -1033,9 +1032,7 @@ impl ObjectStorage for Gcs {
             .iter()
             .filter_map(|path| {
                 let path_str = path.as_ref();
-                if let Some(stripped) =
-                    path_str.strip_prefix(&format!("{}/{}/{}/", stream_name, date, hour))
-                {
+                if let Some(stripped) = path_str.strip_prefix(&prefix) {
                     // Remove trailing slash if present, otherwise use as is
                     let clean_path = stripped.strip_suffix('/').unwrap_or(stripped);
                     Some(clean_path.to_string())

src/storage/localfs.rs

@@ -49,7 +49,7 @@ use crate::{
 use super::{
     ALERTS_ROOT_DIRECTORY, ObjectStorage, ObjectStorageError, ObjectStorageProvider,
     PARSEABLE_ROOT_DIRECTORY, STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY,
-    TOMBSTONE_ROOT_DIRECTORY,
+    TOMBSTONE_ROOT_DIRECTORY, object_storage::stream_prefix,
 };
 
 #[derive(Debug, Clone, clap::Args)]
@@ -491,7 +491,7 @@ impl ObjectStorage for LocalFS {
         stream_name: &str,
         tenant_id: &Option<String>,
     ) -> Result<(), ObjectStorageError> {
-        let path = self.root.join(stream_name);
+        let path = self.root.join(stream_prefix(stream_name, tenant_id));
         let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
 
         let result = fs::remove_dir_all(path).await;
@@ -683,7 +683,7 @@ impl ObjectStorage for LocalFS {
         stream_name: &str,
         tenant_id: &Option<String>,
     ) -> Result<Vec<String>, ObjectStorageError> {
-        let path = self.root.join(stream_name);
+        let path = self.root.join(stream_prefix(stream_name, tenant_id));
         let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
 
         let result = fs::read_dir(&path).await;
@@ -717,9 +717,12 @@ impl ObjectStorage for LocalFS {
         &self,
         stream_name: &str,
         date: &str,
-        _tenant_id: &Option<String>,
+        tenant_id: &Option<String>,
     ) -> Result<Vec<String>, ObjectStorageError> {
-        let path = self.root.join(stream_name).join(date);
+        let path = self
+            .root
+            .join(stream_prefix(stream_name, tenant_id))
+            .join(date);
         let directories = ReadDirStream::new(fs::read_dir(&path).await?);
         let entries: Vec<DirEntry> = directories.try_collect().await?;
         let entries = entries.into_iter().map(dir_name);
@@ -736,9 +739,13 @@ impl ObjectStorage for LocalFS {
         stream_name: &str,
         date: &str,
         hour: &str,
-        _tenant_id: &Option<String>,
+        tenant_id: &Option<String>,
     ) -> Result<Vec<String>, ObjectStorageError> {
-        let path = self.root.join(stream_name).join(date).join(hour);
+        let path = self
+            .root
+            .join(stream_prefix(stream_name, tenant_id))
+            .join(date)
+            .join(hour);
         // Propagate any read_dir errors instead of swallowing them
         let directories = ReadDirStream::new(fs::read_dir(&path).await?);
         let entries: Vec<DirEntry> = directories.try_collect().await?;
@@ -916,3 +923,85 @@ impl From<fs_extra::error::Error> for ObjectStorageError {
         ObjectStorageError::UnhandledError(Box::new(e))
     }
 }
+
+#[cfg(test)]
+mod tenant_path_tests {
+    //! Regression tests for tenant-scoped object paths.
+    //!
+    //! Regression tests for the centralized tenant-scoped path construction
+    //! (`stream_prefix`/`tenant_prefix`) and the backend list/delete paths.
+    //! They cover issues/0001 (retention deletes the wrong prefix) and
+    //! issues/0002 (list APIs ignore the tenant) at the LocalFS layer, without
+    //! needing the global `PARSEABLE` server bootstrap.
+
+    use super::*;
+
+    fn make_dir(root: &Path, rel: &str) {
+        std::fs::create_dir_all(root.join(rel)).unwrap();
+    }
+
+    /// `list_dates` must list the requested tenant's prefix, not the bucket root.
+    #[tokio::test]
+    async fn list_dates_must_be_tenant_scoped() {
+        let tmp = temp_dir::TempDir::new().unwrap();
+        let root = tmp.path().to_path_buf();
+
+        // Same-named stream under two tenants, with distinguishable partitions.
+        make_dir(&root, "t1/logs/date=2024-01-01");
+        make_dir(&root, "logs/date=2024-03-03");
+
+        let store = LocalFS::new(root.clone());
+        let dates = store
+            .list_dates("logs", &Some("t1".to_owned()))
+            .await
+            .unwrap();
+
+        assert_eq!(
+            dates,
+            vec!["date=2024-01-01".to_owned()],
+            "list_dates for tenant t1 returned {dates:?}; it is listing the \
+             tenant-less prefix (see issues/0002)"
+        );
+    }
+
+    /// `stream_prefix` is the single source of truth for tenant path scoping.
+    #[test]
+    fn stream_prefix_is_tenant_scoped() {
+        assert_eq!(stream_prefix("logs", &None), "logs");
+        assert_eq!(stream_prefix("logs", &Some("t1".to_owned())), "t1/logs");
+        // The default-tenant sentinel is treated as "no tenant".
+        assert_eq!(
+            stream_prefix("logs", &Some(DEFAULT_TENANT.to_owned())),
+            "logs"
+        );
+    }
+
+    /// A retention delete for tenant `t1` must delete exactly `t1`'s expired
+    /// partition and must not touch the default tenant's same-named stream.
+    #[tokio::test]
+    async fn retention_delete_is_tenant_scoped() {
+        let tmp = temp_dir::TempDir::new().unwrap();
+        let root = tmp.path().to_path_buf();
+
+        make_dir(&root, "t1/logs/date=2024-01-01");
+        make_dir(&root, "logs/date=2024-01-01");
+
+        let store = LocalFS::new(root.clone());
+        let tenant = Some("t1".to_owned());
+        let date = "date=2024-01-01".to_owned();
+
+        // The path retention now builds (see `src/storage/retention.rs`).
+        let prefix = stream_prefix("logs", &tenant);
+        let path = RelativePathBuf::from_iter([prefix.as_str(), &date]);
+        store.delete_prefix(&path, &tenant).await.unwrap();
+
+        assert!(
+            !root.join("t1/logs/date=2024-01-01").exists(),
+            "tenant t1's expired partition should be deleted"
+        );
+        assert!(
+            root.join("logs/date=2024-01-01").exists(),
+            "the default tenant's same-named stream must not be touched"
+        );
+    }
+}