rust(feat): Improve backup directory names and structure (#326)

tsift · web-flow · commit 2c88ca78a090 · 2025-09-29T17:25:34.000-07:00
diff --git a/rust/crates/sift_stream/src/backup/mod.rs b/rust/crates/sift_stream/src/backup/mod.rs
@@ -33,3 +33,19 @@ enum Message<T> {
     /// Force the backup task to flush its contents to the target data container.
     Flush,
 }
+
+/// Sanitize a name by replacing illegal characters with underscores.
+pub(crate) fn sanitize_name(name: &str) -> String {
+    name.chars()
+        .map(|c| match c {
+            ':' | '/' | '\\' | '*' | '?' | '"' | '<' | '>' | '|' | '.' => '_',
+            _ => {
+                if c.is_whitespace() {
+                    '_'
+                } else {
+                    c
+                }
+            }
+        })
+        .collect()
+}
diff --git a/rust/crates/sift_stream/src/backup/test.rs b/rust/crates/sift_stream/src/backup/test.rs
@@ -1,106 +1,29 @@
 use super::{BackupsManager, DiskBackupPolicy, DiskBackupsManager, InMemoryBackupsManager};
-use crate::SiftChannel;
-use crate::TimeValue;
 use crate::backup::disk::AsyncBackupsManager;
-use hyper_util::rt::TokioIo;
-use sift_connect::grpc::interceptor::AuthInterceptor;
+use crate::{TimeValue, backup::sanitize_name};
 use sift_error::ErrorKind;
 use sift_rs::ingest::v1::{
     IngestWithConfigDataChannelValue, IngestWithConfigDataStreamRequest,
     ingest_with_config_data_channel_value::Type,
 };
-use sift_rs::ingest::v1::{
-    IngestWithConfigDataStreamResponse,
-    ingest_service_server::{IngestService, IngestServiceServer},
-};
 use std::fs;
-use std::sync::{Arc, Mutex};
 use tempdir::TempDir;
-use tonic::transport::{Endpoint, Server, Uri};
-use tonic::{Request, Response, Status};
-use tower::{ServiceBuilder, service_fn};
-
-#[derive(Debug, Clone)]
-struct MockIngestService {
-    captured_data: Arc<Mutex<Vec<IngestWithConfigDataStreamRequest>>>,
-}
-
-impl MockIngestService {
-    fn new() -> Self {
-        Self {
-            captured_data: Arc::new(Mutex::new(Vec::new())),
-        }
-    }
 
-    fn get_captured_data(&self) -> Vec<IngestWithConfigDataStreamRequest> {
-        self.captured_data.lock().unwrap().clone()
+#[test]
+fn test_sanitize_name_with_illegal_chars() {
+    let illegal_chars = vec![
+        ':', '/', '\\', '*', '?', '"', '<', '>', '|', '.', ' ', '\t', '\n', '\r',
+    ];
+    for char in illegal_chars {
+        assert_eq!(sanitize_name(&format!("test{}test", char)), "test_test");
     }
 }
 
-#[tonic::async_trait]
-impl IngestService for MockIngestService {
-    async fn ingest_with_config_data_stream(
-        &self,
-        request: Request<tonic::Streaming<IngestWithConfigDataStreamRequest>>,
-    ) -> Result<Response<IngestWithConfigDataStreamResponse>, Status> {
-        let mut stream = request.into_inner();
-
-        while let Some(data) = stream.message().await? {
-            self.captured_data.lock().unwrap().push(data);
-        }
-
-        Ok(Response::new(IngestWithConfigDataStreamResponse {}))
-    }
-
-    async fn ingest_arbitrary_protobuf_data_stream(
-        &self,
-        _request: Request<
-            tonic::Streaming<sift_rs::ingest::v1::IngestArbitraryProtobufDataStreamRequest>,
-        >,
-    ) -> Result<Response<sift_rs::ingest::v1::IngestArbitraryProtobufDataStreamResponse>, Status>
-    {
-        Err(Status::unimplemented("Not implemented for test"))
-    }
-}
-
-async fn create_mock_grpc_channel_with_service() -> (SiftChannel, Arc<MockIngestService>) {
-    let mock_service = Arc::new(MockIngestService::new());
-    let service_clone = mock_service.clone();
-
-    let (client, server) = tokio::io::duplex(1024);
-
-    tokio::spawn(async move {
-        Server::builder()
-            .add_service(IngestServiceServer::new(service_clone.as_ref().clone()))
-            .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server)))
-            .await
-            .unwrap();
-    });
-
-    let mut client = Some(client);
-    let channel = Endpoint::try_from("http://[::]:50051")
-        .unwrap()
-        .connect_with_connector(service_fn(move |_: Uri| {
-            let client = client.take();
-
-            async move {
-                if let Some(client) = client {
-                    Ok(TokioIo::new(client))
-                } else {
-                    Err(std::io::Error::other("Client already taken"))
-                }
-            }
-        }))
-        .await
-        .unwrap();
-
-    let sift_channel = ServiceBuilder::new()
-        .layer(tonic::service::interceptor(AuthInterceptor {
-            apikey: "test-api-key".to_string(),
-        }))
-        .service(channel);
-
-    (sift_channel, mock_service)
+#[test]
+fn test_sanitize_name_with_legal_chars() {
+    assert_eq!(sanitize_name("test"), "test");
+    assert_eq!(sanitize_name("test_test"), "test_test");
+    assert_eq!(sanitize_name("test-test"), "test-test");
 }
 
 #[tokio::test]
@@ -229,7 +152,7 @@ async fn test_async_backups_manager_retrieve_data_with_graceful_termination() {
         ..Default::default()
     };
     let backup_retry_policy = crate::RetryPolicy::default();
-    let (grpc_channel, mock_service) = create_mock_grpc_channel_with_service().await;
+    let (grpc_channel, mock_service) = crate::test::create_mock_grpc_channel_with_service().await;
 
     let mut backups_manager = AsyncBackupsManager::<IngestWithConfigDataStreamRequest>::new(
         &backups_dir,
@@ -308,7 +231,7 @@ async fn test_async_backups_manager_discard_data_with_graceful_termination() {
         ..Default::default()
     };
     let backup_retry_policy = crate::RetryPolicy::default();
-    let (grpc_channel, mock_service) = create_mock_grpc_channel_with_service().await;
+    let (grpc_channel, mock_service) = crate::test::create_mock_grpc_channel_with_service().await;
 
     let mut backups_manager = AsyncBackupsManager::<IngestWithConfigDataStreamRequest>::new(
         &backups_dir,
diff --git a/rust/crates/sift_stream/src/lib.rs b/rust/crates/sift_stream/src/lib.rs
@@ -423,3 +423,6 @@ pub mod backup;
 pub use backup::DiskBackupPolicy;
 
 pub use sift_connect::grpc::{Credentials, SiftChannel};
+
+#[cfg(test)]
+mod test;
diff --git a/rust/crates/sift_stream/src/stream/builder.rs b/rust/crates/sift_stream/src/stream/builder.rs
@@ -8,6 +8,7 @@ use super::{
 use crate::backup::{
     DiskBackupsManager, InMemoryBackupsManager,
     disk::{AsyncBackupsManager, DiskBackupPolicy},
+    sanitize_name,
 };
 use sift_connect::{Credentials, SiftChannel, SiftChannelBuilder};
 use sift_error::prelude::*;
@@ -325,6 +326,8 @@ impl SiftStreamBuilder<IngestionConfigMode> {
             }
         };
 
+        let asset_name = asset.name.clone();
+
         // Try updating tags or metadata. Update only occurs if either asset_tags or asset_metadata is Some
         Self::update_asset_tags_and_metadata(
             asset,
@@ -376,10 +379,14 @@ impl SiftStreamBuilder<IngestionConfigMode> {
                     retry_policy,
                     disk_backup_policy,
                 } => {
+                    let mut dir_name = sanitize_name(&asset_name);
+                    if let Some(run) = run.as_ref() {
+                        dir_name.push_str(&format!("/{}", sanitize_name(&run.name)));
+                    }
                     policy = Some(retry_policy.clone());
                     let manager = AsyncBackupsManager::new(
-                        &ingestion_config.asset_id,
-                        &ingestion_config.ingestion_config_id,
+                        &dir_name,
+                        &ingestion_config.client_key,
                         disk_backup_policy,
                         retry_policy,
                         channel.clone(),
diff --git a/rust/crates/sift_stream/src/stream/mod.rs b/rust/crates/sift_stream/src/stream/mod.rs
@@ -25,6 +25,9 @@ pub mod time;
 /// in a manner that isn't backwards compatible.
 pub(crate) mod flow;
 
+#[cfg(test)]
+mod test;
+
 /// [SiftStream] is a smart wrapper over an actual gRPC stream that makes it robust and more
 /// ergonomic to work with. Some additional behaviors that [SiftStream] supports are:
 /// - Checkpointing
diff --git a/rust/crates/sift_stream/src/stream/test.rs b/rust/crates/sift_stream/src/stream/test.rs
@@ -0,0 +1,152 @@
+use std::fs;
+
+use crate::TimeValue;
+use crate::backup::DiskBackupPolicy;
+use crate::{
+    ChannelValue, Flow, IngestionConfigForm, RecoveryStrategy, RunForm, SiftStreamBuilder,
+};
+use tempdir::TempDir;
+
+#[tokio::test]
+async fn test_sift_stream_builder_backup_manager_directory_naming_with_run() {
+    let backups_dir = uuid::Uuid::new_v4().to_string();
+
+    let tmp_dir = TempDir::new(&backups_dir).expect("failed to creat tempdir");
+    let tmp_dir_path = tmp_dir.path();
+
+    let ingestion_config = IngestionConfigForm {
+        asset_name: "test_asset".to_string(),
+        client_key: "test_client_key".to_string(),
+        flows: vec![],
+    };
+    let run = RunForm {
+        name: "test_run".to_string(),
+        client_key: "test_client_key".to_string(),
+        description: None,
+        tags: None,
+        metadata: None,
+    };
+
+    let disk_backup_policy = DiskBackupPolicy {
+        backups_dir: Some(tmp_dir_path.to_path_buf()),
+        ..Default::default()
+    };
+    let retry_policy = crate::RetryPolicy::default();
+    let (grpc_channel, _mock_service) = crate::test::create_mock_grpc_channel_with_service().await;
+
+    let mut sift_stream = SiftStreamBuilder::from_channel(grpc_channel)
+        .ingestion_config(ingestion_config)
+        .attach_run(run)
+        .recovery_strategy(RecoveryStrategy::RetryWithBackups {
+            retry_policy,
+            disk_backup_policy,
+        })
+        .build()
+        .await
+        .expect("failed to build sift stream");
+
+    for data in 0..100 {
+        sift_stream
+            .send(Flow::new(
+                "some_flow",
+                TimeValue::now(),
+                &[ChannelValue::new("some_channel", data)],
+            ))
+            .await
+            .expect("failed to send data to backup task");
+    }
+
+    let test_dir = fs::read_dir(tmp_dir_path)
+        .expect("failed to read backups directory")
+        .collect::<Vec<_>>();
+    assert_eq!(test_dir.len(), 1);
+
+    // The first subdirectory should be the asset name.
+    let asset_dir = test_dir[0].as_ref().expect("failed to get file");
+    assert!(asset_dir.path().is_dir(), "expected file to be a directory");
+
+    let asset_dir_path = asset_dir.path();
+    let asset_dir_file_name = asset_dir_path.file_name().expect("failed to get file name");
+    assert_eq!(asset_dir_file_name, "test_asset");
+
+    // The next subdirectory in the asset directory should be the run name.
+    let asset_dir_contents = fs::read_dir(asset_dir_path)
+        .expect("failed to read asset directory")
+        .collect::<Vec<_>>();
+    assert_eq!(asset_dir_contents.len(), 1);
+
+    let run_dir = asset_dir_contents[0].as_ref().expect("failed to get file");
+    assert!(run_dir.path().is_dir(), "expected file to be a directory");
+
+    let run_dir_path = run_dir.path();
+    let run_dir_name = run_dir_path.file_name().expect("failed to get file name");
+    assert_eq!(run_dir_name, "test_run");
+}
+
+#[tokio::test]
+async fn test_sift_stream_builder_backup_manager_directory_naming_no_run() {
+    let backups_dir = uuid::Uuid::new_v4().to_string();
+
+    let tmp_dir = TempDir::new(&backups_dir).expect("failed to creat tempdir");
+    let tmp_dir_path = tmp_dir.path();
+
+    let ingestion_config = IngestionConfigForm {
+        asset_name: "test_asset".to_string(),
+        client_key: "test_client_key".to_string(),
+        flows: vec![],
+    };
+    let disk_backup_policy = DiskBackupPolicy {
+        backups_dir: Some(tmp_dir_path.to_path_buf()),
+        ..Default::default()
+    };
+    let retry_policy = crate::RetryPolicy::default();
+    let (grpc_channel, _mock_service) = crate::test::create_mock_grpc_channel_with_service().await;
+
+    let mut sift_stream = SiftStreamBuilder::from_channel(grpc_channel)
+        .ingestion_config(ingestion_config)
+        .recovery_strategy(RecoveryStrategy::RetryWithBackups {
+            retry_policy,
+            disk_backup_policy,
+        })
+        .build()
+        .await
+        .expect("failed to build sift stream");
+
+    for data in 0..100 {
+        sift_stream
+            .send(Flow::new(
+                "some_flow",
+                TimeValue::now(),
+                &[ChannelValue::new("some_channel", data)],
+            ))
+            .await
+            .expect("failed to send data to backup task");
+    }
+
+    let test_dir = fs::read_dir(tmp_dir_path)
+        .expect("failed to read backups directory")
+        .collect::<Vec<_>>();
+    assert_eq!(test_dir.len(), 1);
+
+    // The first subdirectory should be the asset name.
+    let asset_dir = test_dir[0].as_ref().expect("failed to get file");
+    assert!(asset_dir.path().is_dir(), "expected file to be a directory");
+
+    let asset_dir_path = asset_dir.path();
+    let asset_dir_file_name = asset_dir_path.file_name().expect("failed to get file name");
+    assert_eq!(asset_dir_file_name, "test_asset");
+
+    // Since there was no run provided, there are no subdirectories in the asset directory.
+    let asset_dir_contents = fs::read_dir(asset_dir_path)
+        .expect("failed to read asset directory")
+        .collect::<Vec<_>>();
+    assert_eq!(asset_dir_contents.len(), 1);
+    assert!(
+        asset_dir_contents[0]
+            .as_ref()
+            .expect("failed to get file")
+            .path()
+            .is_file(),
+        "expected to be a file",
+    );
+}
diff --git a/rust/crates/sift_stream/src/test.rs b/rust/crates/sift_stream/src/test.rs
