Retention deletes the wrong object prefix for named tenants (cross-tenant data loss) #1812

for date in dates_to_delete {
    let path = RelativePathBuf::from_iter([&stream_name, &date]);
    delete_tasks.push(async move {
        PARSEABLE
            .storage
            .get_object_store()
            .delete_prefix(&path, tenant_id)
            .await
    });
}

async fn delete_prefix(&self, path: &RelativePath, tenant_id: &Option<String>) -> ... {
    self._delete_prefix(path.as_ref(), tenant_id).await?;
    ...
}

if let Some(tenant) = tenant_id
    && !tenant.eq(DEFAULT_TENANT)
{
    format!("{tenant}/{stream_name}/{file_suffix}")
} else {
    format!("{stream_name}/{file_suffix}")
}

#[cfg(test)]
mod tenant_path_tests {
    //! Regression tests for tenant-scoped object paths.
    //! Reproduces retention deleting the wrong prefix, and list APIs ignoring
    //! the tenant, at the LocalFS backend layer.

    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"
        );
    }

    /// A retention delete for tenant `t1` must not touch the default tenant's
    /// same-named stream.
    #[tokio::test]
    async fn retention_delete_must_not_touch_other_tenants() {
        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());

        // Exact path construction from `src/storage/retention.rs` (action::delete).
        let stream_name = "logs".to_owned();
        let date = "date=2024-01-01".to_owned();
        let path = RelativePathBuf::from_iter([&stream_name, &date]);

        // Retention runs for tenant `t1`.
        store
            .delete_prefix(&path, &Some("t1".to_owned()))
            .await
            .unwrap();

        assert!(
            root.join("t1/logs/date=2024-01-01").exists(),
            "tenant t1's own data should still be handled correctly"
        );
        assert!(
            root.join("logs/date=2024-01-01").exists(),
            "BUG: a retention run for tenant t1 deleted the DEFAULT tenant's \
             logs/date=2024-01-01"
        );
    }
}

pub fn stream_prefix(stream: &str, tenant_id: &Option<String>) -> RelativePathBuf {
    match tenant_id {
        Some(t) if !t.eq(DEFAULT_TENANT) => RelativePathBuf::from_iter([t, stream]),
        _ => RelativePathBuf::from(stream),
    }
}

let path = stream_prefix(&stream_name, tenant_id).join(&date);