rust(bugfix): Prefer streaming data on channel full instead of erroring (#363)

* Log warning when backup data channel is full instead of returning to caller.
* Bump rc version and add changelog entry.

Author: tsift

Cargo.toml

@@ -12,7 +12,7 @@ members = [
 
 [workspace.package]
 authors = ["Sift Software Engineers <engineering@siftstack.com>"]
-version = "0.7.0-rc.1"
+version = "0.7.0-rc.2"
 edition = "2024"
 categories = ["aerospace", "science::robotics"]
 homepage = "https://github.com/sift-stack/sift"
@@ -26,10 +26,10 @@ chrono = { version = "0.4.39", default-features = false, features = ["clock"] }
 pbjson-types = "^0.7"
 tonic = { version = "^0.12", features = ["gzip"] }
 
-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_connect = { version = "0.7.0-rc.2", path = "rust/crates/sift_connect" }
+sift_rs = { version = "0.7.0-rc.2", path = "rust/crates/sift_rs" }
+sift_error = { version = "0.7.0-rc.2", path = "rust/crates/sift_error" }
+sift_stream = { version = "0.7.0-rc.2", path = "rust/crates/sift_stream" }
 
 sift_stream_bindings = { version = "0.1.0", path = "rust/crates/sift_stream_bindings" }
 

rust/CHANGELOG.md

@@ -3,6 +3,15 @@ All notable changes to this project will be documented in this file.
 
 This project adheres to [Semantic Versioning](http://semver.org/).
 
+## [v0.7.0-rc.2] - November 3, 2025
+### What's New
+#### Improvements For Constrained Environments
+Compression support has been added for streaming data into Sift. This can be useful in low-bandwidth or network
+constrained environments. Since compression does add overhead to streaming, it is not recommended for high
+throughput streaming systems.
+
+Additionally, changes have been made to ensure data is streamed to Sift even when writing backup files lags
+behind ingestion.
 
 ## [v0.7.0-rc.1] - October 24, 2025
 ### What's New

rust/crates/sift_stream/README.md

@@ -111,19 +111,23 @@ error handling, and shutdown processes.
 - **backup_tx/backup_rx**: Bounded channel for backup data messages
 
 **IMPORTANT**:
-Data reliability is among the most important requirements of sift-stream, thus if the data channel used for backups
-becomes full, an error is returned to the caller. This is in contrast to the data channel used for the primary
-ingestion into Sift -- if this channel becomes full, the oldest data will be removed in favor of streaming newer
-data. The data removed during this process will have been backed up to disk and will be re-ingested at the next
-checkpoint.
+Data reliability is among the most important goals of sift-stream, however if the backup system falls
+behind (slow disk writes), a warning will be emitted but data will still be sent for ingestion.
+Streaming data in this situation is preferred over preventing it entirely. It is **highly** recommended
+however that write speeds be sufficiently high for the given data stream being backed up to ensure data
+is reliably backed up.
+
+In contrast, if the primary ingestion channel becomes full, the oldest data will be removed in favor of
+streaming newer data. The data removed during this process will be forwarded to the backup system and
+will be re-ingested at the next checkpoint.
 
 ### Control Channel
-- **control_tx/control_rx**: Broadcast channel (1,024 capacity) for low-frequency control messages
+- **control_tx/control_rx**: Broadcast channel for low-frequency control messages
 
 ### Channel Capacities
 
 The default capacities are as follows:
-- **data**: 10,240
+- **data**: 102,400
 - **control**: 1024
 
 These can be configured however, based on individual streaming needs.

rust/crates/sift_stream/src/stream/builder.rs

@@ -56,7 +56,8 @@ pub struct SiftStreamBuilder<C> {
     asset_tags: Option<Vec<String>>,
     asset_metadata: Option<Vec<MetadataValue>>,
     control_channel_capacity: usize,
-    data_channel_capacity: usize,
+    ingestion_data_channel_capacity: usize,
+    backup_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.
@@ -136,10 +137,17 @@ where
         self
     }
 
-    /// Sets the data channel capacity. See the [top-level documentation](crate#checkpoints)
+    /// Sets the ingestion data channel capacity. See the [top-level documentation](crate#checkpoints)
     /// for further details.
-    pub fn data_channel_capacity(mut self, capacity: usize) -> SiftStreamBuilder<C> {
-        self.data_channel_capacity = capacity;
+    pub fn ingestion_data_channel_capacity(mut self, capacity: usize) -> SiftStreamBuilder<C> {
+        self.ingestion_data_channel_capacity = capacity;
+        self
+    }
+
+    /// Sets the backup data channel capacity. See the [top-level documentation](crate#checkpoints)
+    /// for further details.
+    pub fn backup_data_channel_capacity(mut self, capacity: usize) -> SiftStreamBuilder<C> {
+        self.backup_data_channel_capacity = capacity;
         self
     }
 
@@ -222,7 +230,8 @@ impl SiftStreamBuilder<IngestionConfigMode> {
             asset_tags: None,
             asset_metadata: None,
             control_channel_capacity: CONTROL_CHANNEL_CAPACITY,
-            data_channel_capacity: DATA_CHANNEL_CAPACITY,
+            ingestion_data_channel_capacity: DATA_CHANNEL_CAPACITY,
+            backup_data_channel_capacity: DATA_CHANNEL_CAPACITY,
         }
     }
 
@@ -242,7 +251,8 @@ impl SiftStreamBuilder<IngestionConfigMode> {
             asset_tags: None,
             asset_metadata: None,
             control_channel_capacity: CONTROL_CHANNEL_CAPACITY,
-            data_channel_capacity: DATA_CHANNEL_CAPACITY,
+            ingestion_data_channel_capacity: DATA_CHANNEL_CAPACITY,
+            backup_data_channel_capacity: DATA_CHANNEL_CAPACITY,
         }
     }
 
@@ -360,7 +370,8 @@ impl SiftStreamBuilder<IngestionConfigMode> {
             enable_compression_for_ingestion,
             recovery_config,
             control_channel_capacity: self.control_channel_capacity,
-            data_channel_capacity: self.data_channel_capacity,
+            ingestion_data_channel_capacity: self.ingestion_data_channel_capacity,
+            backup_data_channel_capacity: self.backup_data_channel_capacity,
         };
 
         SiftStream::<IngestionConfigMode>::new(ingestion_config, flows, run, task_config, metrics)

rust/crates/sift_stream/src/stream/mode/ingestion_config.rs

@@ -132,17 +132,15 @@ impl SiftStream<IngestionConfigMode> {
     /// to register the flow before calling [SiftStream::send]; otherwise users should monitor the
     /// Sift DLQ either in the Sift UI or Sift API to ensure successful transmission.
     ///
-    /// In the case where the underlying error was closed due to an error, this method will invoke
-    /// the configured [RetryPolicy] to attempt to reconnect and resend data to Sift. If the amount
-    /// of retry attempts exceeds the maximum configured, then an [ErrorKind::RetriesExhausted] is
-    /// returned. If backups are enabled, then all messages since the last successful checkpoint
-    /// will be reingested.
+    /// When "sending" messages, first the message will sent to the backup system. This system
+    /// is used to backup data to disk until the data is confirmed received by Sift. If streaming
+    /// encounters errors, the backed up data will be re-ingested ensuring all data is received
+    /// by Sift.
     ///
-    /// Lastly, if the underlying stream was gracefully closed due to a checkpoint, this method
-    /// will automatically establish a new connection.
+    /// If the backup system has fallen behind and the backup queue/channel is full, it will
+    /// proceed to sending the message to Sift.
     ///
-    /// TODO: To preserve the API, this function is remaining async even though it no longer has
-    /// any `await`s. This should be changed to a blocking function in the next major version.
+    /// This ensures data is sent to Sift even if the backup system is lagging.
     pub async fn send(&mut self, message: Flow) -> Result<()> {
         self.metrics.messages_received.increment();
 
@@ -227,15 +225,18 @@ impl SiftStream<IngestionConfigMode> {
             dropped_for_ingestion: false,
         };
 
-        // Send the message for backup first. If this fails, return an error.
+        // Send the message for backup first. If this fails, log an error and continue.
         //
-        // Failure to backup leads to data loss and should be treated as very critical.
-        self.mode
-            .stream_system
-            .backup_tx
-            .try_send(data_msg.clone())
-            .map_err(|e| Error::new(ErrorKind::StreamError, e))
-            .context("failed to send data to backup task system")?;
+        // Failure to backup can lead to data loss though it is preferable to attempt
+        // to stream the message to Sift rather than return the error and prevent both.
+        if let Err(e) = self.mode.stream_system.backup_tx.try_send(data_msg.clone()) {
+            #[cfg(feature = "tracing")]
+            tracing::warn!(
+                sift_stream_id = self.mode.sift_stream_id.to_string(),
+                "failed to send data to backup system, data will still be streamed to Sift: {e}"
+            );
+        }
+
         self.metrics.messages_sent_to_backup.increment();
 
         // Send the message for ingestion.
@@ -267,7 +268,7 @@ impl SiftStream<IngestionConfigMode> {
             }
             Err(e) => Err(Error::new_msg(
                 ErrorKind::StreamError,
-                format!("queueing data for ingestion failed: {}", e),
+                format!("queueing data for ingestion failed: {e}"),
             )),
         }
     }

rust/crates/sift_stream/src/stream/tasks.rs

@@ -16,7 +16,7 @@ use tokio::{sync::broadcast, task::JoinHandle, time::Instant};
 use uuid::Uuid;
 
 /// Capacity for the data channel.
-pub(crate) const DATA_CHANNEL_CAPACITY: usize = 1024 * 10;
+pub(crate) const DATA_CHANNEL_CAPACITY: usize = 1024 * 100;
 
 /// Capacity for the control channel.
 pub(crate) const CONTROL_CHANNEL_CAPACITY: usize = 1024;
@@ -66,7 +66,8 @@ pub(crate) struct TaskConfig {
     pub(crate) enable_compression_for_ingestion: bool,
     pub(crate) recovery_config: RecoveryConfig,
     pub(crate) control_channel_capacity: usize,
-    pub(crate) data_channel_capacity: usize,
+    pub(crate) ingestion_data_channel_capacity: usize,
+    pub(crate) backup_data_channel_capacity: usize,
 }
 
 /// Data message with stream ID for routing
@@ -92,8 +93,9 @@ pub(crate) fn start_tasks(config: TaskConfig) -> Result<StreamSystem> {
     let (control_tx, _control_rx) = broadcast::channel(config.control_channel_capacity);
 
     // Create data channel for high-frequency data messages
-    let (ingestion_tx, ingestion_rx) = async_channel::bounded(config.data_channel_capacity);
-    let (backup_tx, backup_rx) = async_channel::bounded(config.data_channel_capacity);
+    let (ingestion_tx, ingestion_rx) =
+        async_channel::bounded(config.ingestion_data_channel_capacity);
+    let (backup_tx, backup_rx) = async_channel::bounded(config.backup_data_channel_capacity);
 
     // Clone the sender for each task
     let backup_control_tx = control_tx.clone();
@@ -498,7 +500,8 @@ mod tests {
             checkpoint_interval,
             enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
-            data_channel_capacity: 128,
+            ingestion_data_channel_capacity: 128,
+            backup_data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
                 retry_policy: RetryPolicy::default(),
                 backups_enabled: true,
@@ -557,7 +560,8 @@ mod tests {
             checkpoint_interval,
             enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
-            data_channel_capacity: 128,
+            ingestion_data_channel_capacity: 128,
+            backup_data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
                 retry_policy: RetryPolicy::default(),
                 backups_enabled: true,
@@ -610,7 +614,8 @@ mod tests {
             checkpoint_interval,
             enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
-            data_channel_capacity: 128,
+            ingestion_data_channel_capacity: 128,
+            backup_data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
                 retry_policy: RetryPolicy::default(),
                 backups_enabled: true,
@@ -673,7 +678,8 @@ mod tests {
             checkpoint_interval: Duration::from_secs(60),
             enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
-            data_channel_capacity: 128,
+            ingestion_data_channel_capacity: 128,
+            backup_data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
                 retry_policy: RetryPolicy::default(),
                 backups_enabled: true,
@@ -750,7 +756,8 @@ mod tests {
             checkpoint_interval,
             enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
-            data_channel_capacity: 128,
+            ingestion_data_channel_capacity: 128,
+            backup_data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
                 retry_policy: RetryPolicy {
                     max_attempts: 3,
@@ -843,7 +850,8 @@ mod tests {
             checkpoint_interval,
             enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
-            data_channel_capacity: 128,
+            ingestion_data_channel_capacity: 128,
+            backup_data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
                 retry_policy: RetryPolicy::default(),
                 backups_enabled: true,
@@ -914,7 +922,8 @@ mod tests {
             checkpoint_interval,
             enable_compression_for_ingestion: false,
             control_channel_capacity: 128,
-            data_channel_capacity: 128,
+            ingestion_data_channel_capacity: 128,
+            backup_data_channel_capacity: 128,
             recovery_config: RecoveryConfig {
                 retry_policy: RetryPolicy::default(),
                 backups_enabled: true,