rust(feat): Add gzip compression support in sift-stream ingestion (#360)

tsift · web-flow · commit 68b08c86c506 · 2025-11-03T09:45:31.000-08:00
diff --git a/Cargo.toml b/Cargo.toml
@@ -12,7 +12,7 @@ members = [
 
 [workspace.package]
 authors = ["Sift Software Engineers <engineering@siftstack.com>"]
-version = "0.6.0"
+version = "0.7.0-rc.1"
 edition = "2024"
 categories = ["aerospace", "science::robotics"]
 homepage = "https://github.com/sift-stack/sift"
@@ -24,12 +24,12 @@ license = "MIT"
 [workspace.dependencies]
 chrono = { version = "0.4.39", default-features = false, features = ["clock"] }
 pbjson-types = "^0.7"
-tonic = { version = "^0.12" }
+tonic = { version = "^0.12", features = ["gzip"] }
 
-sift_connect = { version = "0.6.0", path = "rust/crates/sift_connect" }
-sift_rs = { version = "0.6.0", path = "rust/crates/sift_rs" }
-sift_error = { version = "0.6.0", path = "rust/crates/sift_error" }
-sift_stream = { version = "0.6.0", path = "rust/crates/sift_stream" }
+sift_connect = { version = "0.7.0-rc.1", path = "rust/crates/sift_connect" }
+sift_rs = { version = "0.7.0-rc.1", path = "rust/crates/sift_rs" }
+sift_error = { version = "0.7.0-rc.1", path = "rust/crates/sift_error" }
+sift_stream = { version = "0.7.0-rc.1", path = "rust/crates/sift_stream" }
 
 sift_stream_bindings = { version = "0.1.0", path = "rust/crates/sift_stream_bindings" }
 
diff --git a/rust/crates/sift_rs/src/lib.rs b/rust/crates/sift_rs/src/lib.rs
@@ -50,3 +50,4 @@ pub use r#gen::sift::*;
 pub mod wrappers;
 
 pub use sift_connect::{Credentials, SiftChannel, SiftChannelBuilder};
+pub use tonic::codec::CompressionEncoding;
diff --git a/rust/crates/sift_stream/src/backup/disk/async_manager.rs b/rust/crates/sift_stream/src/backup/disk/async_manager.rs
@@ -8,6 +8,7 @@ use async_channel::{Receiver, Sender};
 use chrono::Utc;
 use prost::Message as PbMessage;
 use sift_error::prelude::*;
+use sift_rs::CompressionEncoding;
 use sift_rs::{SiftChannel, ingest::v1::ingest_service_client::IngestServiceClient};
 use std::time::Duration;
 use std::{
@@ -407,6 +408,7 @@ pub(crate) struct BackupIngestTask {
     control_rx: broadcast::Receiver<ControlMessage>,
     to_reingest_rx: Receiver<PathBuf>,
     to_reingest_tx: Sender<PathBuf>,
+    enable_compression_for_ingestion: bool,
     grpc_channel: SiftChannel,
     retry_policy: RetryPolicy,
     retain_backups: bool,
@@ -417,6 +419,7 @@ impl BackupIngestTask {
     pub(crate) fn new(
         control_rx: broadcast::Receiver<ControlMessage>,
         grpc_channel: SiftChannel,
+        enable_compression_for_ingestion: bool,
         retry_policy: RetryPolicy,
         retain_backups: bool,
         metrics: Arc<SiftStreamMetrics>,
@@ -426,6 +429,7 @@ impl BackupIngestTask {
             control_rx,
             to_reingest_rx,
             to_reingest_tx,
+            enable_compression_for_ingestion,
             grpc_channel,
             retry_policy,
             retain_backups,
@@ -442,6 +446,7 @@ impl BackupIngestTask {
             self.to_reingest_rx.clone(),
             self.grpc_channel.clone(),
             self.retry_policy,
+            self.enable_compression_for_ingestion,
             self.retain_backups,
             self.metrics.clone(),
         ));
@@ -493,11 +498,19 @@ impl BackupIngestTask {
         to_reingest_rx: Receiver<PathBuf>,
         grpc_channel: SiftChannel,
         retry_policy: RetryPolicy,
+        enable_compression_for_ingestion: bool,
         retain_backups: bool,
         metrics: Arc<SiftStreamMetrics>,
     ) -> Result<()> {
         let mut client = IngestServiceClient::new(grpc_channel);
 
+        // If compression is enabled, add the compression codecs to the client.
+        if enable_compression_for_ingestion {
+            client = client
+                .send_compressed(CompressionEncoding::Gzip)
+                .accept_compressed(CompressionEncoding::Gzip);
+        }
+
         while let Ok(backup_file_path) = to_reingest_rx.recv().await {
             metrics
                 .backups
@@ -1578,6 +1591,7 @@ mod test {
         let (to_reingest_tx, to_reingest_rx) = async_channel::bounded(1024);
         let retry_policy = RetryPolicy::default();
         let retain_backups = false;
+        let enable_compression_for_ingestion = false;
         let (grpc_channel, mock_service) =
             crate::test::create_mock_grpc_channel_with_service().await;
         let metrics = Arc::new(SiftStreamMetrics::default());
@@ -1595,6 +1609,7 @@ mod test {
                 to_reingest_rx,
                 grpc_channel,
                 retry_policy,
+                enable_compression_for_ingestion,
                 retain_backups,
                 metrics
             )
@@ -1637,6 +1652,7 @@ mod test {
         let (to_reingest_tx, to_reingest_rx) = async_channel::bounded(1024);
         let retry_policy = RetryPolicy::default();
         let retain_backups = true;
+        let enable_compression_for_ingestion = false;
         let (grpc_channel, mock_service) =
             crate::test::create_mock_grpc_channel_with_service().await;
         let metrics = Arc::new(SiftStreamMetrics::default());
@@ -1654,6 +1670,7 @@ mod test {
                 to_reingest_rx,
                 grpc_channel,
                 retry_policy,
+                enable_compression_for_ingestion,
                 retain_backups,
                 metrics
             )
@@ -1701,6 +1718,7 @@ mod test {
             backoff_multiplier: 5,
         };
         let retain_backups = false;
+        let enable_compression_for_ingestion = false;
         let (grpc_channel, mock_service) =
             crate::test::create_mock_grpc_channel_with_service().await;
         let metrics = Arc::new(SiftStreamMetrics::default());
@@ -1722,6 +1740,7 @@ mod test {
                 to_reingest_rx,
                 grpc_channel,
                 retry_policy,
+                enable_compression_for_ingestion,
                 retain_backups,
                 metrics.clone()
             )
@@ -1776,6 +1795,7 @@ mod test {
             backoff_multiplier: 5,
         };
         let retain_backups = false;
+        let enable_compression_for_ingestion = false;
         let (grpc_channel, mock_service) =
             crate::test::create_mock_grpc_channel_with_service().await;
         let metrics = Arc::new(SiftStreamMetrics::default());
@@ -1797,6 +1817,7 @@ mod test {
                 to_reingest_rx,
                 grpc_channel,
                 retry_policy,
+                enable_compression_for_ingestion,
                 retain_backups,
                 metrics.clone()
             )
@@ -1846,13 +1867,15 @@ mod test {
         // Create the re-ingestion task.
         let retry_policy = RetryPolicy::default();
         let retain_backups = false;
+        let enable_compression_for_ingestion = false;
         let (grpc_channel, mock_service) =
             crate::test::create_mock_grpc_channel_with_service().await;
         let metrics = Arc::new(SiftStreamMetrics::default());
 
         let reingest_task = BackupIngestTask::new(
             control_rx,
             grpc_channel,
+            enable_compression_for_ingestion,
             retry_policy,
             retain_backups,
             metrics,
@@ -1926,13 +1949,15 @@ mod test {
         // Create the re-ingestion task.
         let retry_policy = RetryPolicy::default();
         let retain_backups = false;
+        let enable_compression_for_ingestion = false;
         let (grpc_channel, _mock_service) =
             crate::test::create_mock_grpc_channel_with_service().await;
         let metrics = Arc::new(SiftStreamMetrics::default());
 
         let reingest_task = BackupIngestTask::new(
             control_rx,
             grpc_channel,
+            enable_compression_for_ingestion,
             retry_policy,
             retain_backups,
             metrics,
diff --git a/rust/crates/sift_stream/src/stream/builder.rs b/rust/crates/sift_stream/src/stream/builder.rs
@@ -57,6 +57,7 @@ pub struct SiftStreamBuilder<C> {
     asset_metadata: Option<Vec<MetadataValue>>,
     control_channel_capacity: usize,
     data_channel_capacity: usize,
+    enable_compression_for_ingestion: bool,
 
     // Either `run` or `run_id`. If both are provided then the `run_id` will be prioritized.
     run: Option<RunForm>,
@@ -167,6 +168,17 @@ where
         self
     }
 
+    /// Sets whether compression is enabled.
+    ///
+    /// Currently only gzip is supported.
+    ///
+    /// WARNING: Compression adds additional overhead both on the client and server, so can reduce
+    /// the overall throughput of a stream. It is not recommended to enable compression by default.
+    pub fn enable_compression_for_ingestion(mut self, enable: bool) -> SiftStreamBuilder<C> {
+        self.enable_compression_for_ingestion = enable;
+        self
+    }
+
     /// Disables TLS. Useful for testing. This is ignored if [SiftStreamBuilder::from_channel] is
     /// used to initialize the builder.
     pub fn disable_tls(mut self) -> SiftStreamBuilder<C> {
@@ -200,6 +212,7 @@ impl SiftStreamBuilder<IngestionConfigMode> {
             credentials: Some(credentials),
             channel: None,
             enable_tls: true,
+            enable_compression_for_ingestion: false,
             ingestion_config: None,
             run: None,
             run_id: None,
@@ -219,6 +232,7 @@ impl SiftStreamBuilder<IngestionConfigMode> {
             credentials: None,
             channel: Some(channel),
             enable_tls: true,
+            enable_compression_for_ingestion: false,
             ingestion_config: None,
             run: None,
             run_id: None,
@@ -240,6 +254,7 @@ impl SiftStreamBuilder<IngestionConfigMode> {
             channel: grpc_channel,
             credentials,
             enable_tls,
+            enable_compression_for_ingestion,
             ingestion_config,
             recovery_strategy,
             run,
@@ -342,6 +357,7 @@ impl SiftStreamBuilder<IngestionConfigMode> {
             grpc_channel: channel,
             metrics: metrics.clone(),
             checkpoint_interval,
+            enable_compression_for_ingestion,
             recovery_config,
             control_channel_capacity: self.control_channel_capacity,
             data_channel_capacity: self.data_channel_capacity,
diff --git a/rust/crates/sift_stream/src/stream/tasks.rs b/rust/crates/sift_stream/src/stream/tasks.rs
@@ -7,8 +7,9 @@ use crate::{
 use async_channel;
 use sift_connect::SiftChannel;
 use sift_error::prelude::*;
-use sift_rs::ingest::v1::{
-    IngestWithConfigDataStreamRequest, ingest_service_client::IngestServiceClient,
+use sift_rs::{
+    CompressionEncoding,
+    ingest::v1::{IngestWithConfigDataStreamRequest, ingest_service_client::IngestServiceClient},
 };
 use std::{path::PathBuf, pin::Pin, sync::Arc, time::Duration};
 use tokio::{sync::broadcast, task::JoinHandle, time::Instant};
@@ -47,31 +48,32 @@ pub(crate) enum ControlMessage {
 }
 
 #[derive(Clone)]
-pub struct RecoveryConfig {
-    pub retry_policy: RetryPolicy,
-    pub backups_enabled: bool,
-    pub backups_directory: String,
-    pub backups_prefix: String,
-    pub backup_policy: DiskBackupPolicy,
+pub(crate) struct RecoveryConfig {
+    pub(crate) retry_policy: RetryPolicy,
+    pub(crate) backups_enabled: bool,
+    pub(crate) backups_directory: String,
+    pub(crate) backups_prefix: String,
+    pub(crate) backup_policy: DiskBackupPolicy,
 }
 
 /// Configuration for the task-based SiftStream
 #[derive(Clone)]
-pub struct TaskConfig {
-    pub sift_stream_id: Uuid,
-    pub grpc_channel: SiftChannel,
-    pub metrics: Arc<SiftStreamMetrics>,
-    pub checkpoint_interval: Duration,
-    pub recovery_config: RecoveryConfig,
-    pub control_channel_capacity: usize,
-    pub data_channel_capacity: usize,
+pub(crate) struct TaskConfig {
+    pub(crate) sift_stream_id: Uuid,
+    pub(crate) grpc_channel: SiftChannel,
+    pub(crate) metrics: Arc<SiftStreamMetrics>,
+    pub(crate) checkpoint_interval: Duration,
+    pub(crate) enable_compression_for_ingestion: bool,
+    pub(crate) recovery_config: RecoveryConfig,
+    pub(crate) control_channel_capacity: usize,
+    pub(crate) data_channel_capacity: usize,
 }
 
 /// Data message with stream ID for routing
 #[derive(Debug, Clone)]
-pub struct DataMessage {
-    pub request: IngestWithConfigDataStreamRequest,
-    pub dropped_for_ingestion: bool,
+pub(crate) struct DataMessage {
+    pub(crate) request: IngestWithConfigDataStreamRequest,
+    pub(crate) dropped_for_ingestion: bool,
 }
 
 /// Handles for the three main tasks
@@ -150,6 +152,7 @@ pub(crate) fn start_tasks(config: TaskConfig) -> Result<StreamSystem> {
     let reingestion_task = BackupIngestTask::new(
         reingestion_control_tx.subscribe(),
         reingestion_config.grpc_channel,
+        reingestion_config.enable_compression_for_ingestion,
         reingest_retry_policy,
         reingestion_config
             .recovery_config
@@ -233,6 +236,14 @@ impl IngestionTask {
                 // any race conditions in that task being polled for the first time and other
                 // events occurring in the system.
                 let mut client = IngestServiceClient::new(self.config.grpc_channel.clone());
+
+                // If compression is enabled, add the compression codecs to the client.
+                if self.config.enable_compression_for_ingestion {
+                    client = client
+                        .send_compressed(CompressionEncoding::Gzip)
+                        .accept_compressed(CompressionEncoding::Gzip);
+                }
+
                 let data_stream = DataStream::new(
                     self.data_rx.clone(),
                     self.control_tx.clone(),
@@ -485,6 +496,7 @@ mod tests {
             grpc_channel,
             metrics: metrics.clone(),
             checkpoint_interval,
+            enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
             data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
@@ -543,6 +555,7 @@ mod tests {
             grpc_channel,
             metrics: metrics.clone(),
             checkpoint_interval,
+            enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
             data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
@@ -595,6 +608,7 @@ mod tests {
             grpc_channel,
             metrics: metrics.clone(),
             checkpoint_interval,
+            enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
             data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
@@ -657,6 +671,7 @@ mod tests {
             grpc_channel,
             metrics: metrics.clone(),
             checkpoint_interval: Duration::from_secs(60),
+            enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
             data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
@@ -733,6 +748,7 @@ mod tests {
             grpc_channel,
             metrics: metrics.clone(),
             checkpoint_interval,
+            enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
             data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
@@ -825,6 +841,7 @@ mod tests {
             grpc_channel,
             metrics: metrics.clone(),
             checkpoint_interval,
+            enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
             data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
@@ -895,6 +912,7 @@ mod tests {
             grpc_channel,
             metrics: metrics.clone(),
             checkpoint_interval,
+            enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
             data_channel_capacity: 128,
             recovery_config: RecoveryConfig {

Original file line number	Diff line number	Diff line change
`@@ -50,3 +50,4 @@ pub use r#gen::sift::*;`
`50`	`50`	`pub mod wrappers;`
`51`	`51`
`52`	`52`	`pub use sift_connect::{Credentials, SiftChannel, SiftChannelBuilder};`
	`53`	`+pub use tonic::codec::CompressionEncoding;`