rust(feat/bug): Improve add flow config, prevent send_impl error (#409)

Author: tsift

rust/crates/sift_error/src/lib.rs

@@ -98,6 +98,8 @@ impl Error {
 /// Various categories of errors that can occur throughout Sift crates.
 #[derive(Debug, PartialEq, Copy, Clone)]
 pub enum ErrorKind {
+    /// Indicates that the error is due to a resource already existing.
+    AlreadyExistsError,
     /// Indicates user-error having to do with bad arguments.
     ArgumentValidationError,
     /// Indicates that the program is unable to grab credentials from a user's `sift.toml` file.
@@ -194,6 +196,7 @@ where
 impl fmt::Display for ErrorKind {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
+            Self::AlreadyExistsError => write!(f, "AlreadyExistsError"),
             Self::GrpcConnectError => write!(f, "GrpcConnectError"),
             Self::RetriesExhausted => write!(f, "RetriesExhausted"),
             Self::RetrieveAssetError => write!(f, "RetrieveAssetError"),

rust/crates/sift_rs/src/wrappers/ingestion_configs.rs

@@ -129,7 +129,13 @@ impl IngestionConfigServiceWrapper for IngestionConfigServiceImpl {
                 flows: configs.into(),
             })
             .await
-            .map_err(|e| Error::new(ErrorKind::CreateFlowError, e))?;
+            .map_err(|e| {
+                if e.code() == tonic::Code::AlreadyExists {
+                    Error::new(ErrorKind::AlreadyExistsError, e)
+                } else {
+                    Error::new(ErrorKind::CreateFlowError, e)
+                }
+            })?;
         Ok(())
     }
 

rust/crates/sift_stream/Cargo.toml

@@ -16,6 +16,7 @@ features = ["metrics-unstable"]
 rustdoc-args = ["--cfg", "docsrs"]
 
 [dependencies]
+futures = { version = "0.3", default-features = false, features = ["alloc"] }
 sift_connect = { workspace = true }
 sift_error = { workspace = true }
 sift_rs = { workspace = true }

rust/crates/sift_stream/src/metrics/mod.rs

@@ -118,6 +118,8 @@ pub struct SiftStreamMetricsSnapshot {
     pub messages_sent_to_backup: u64,
     /// Total messages dropped for ingestion
     pub old_messages_dropped_for_ingestion: u64,
+    /// Total messages dropped for ingestion that failed to be added to backup
+    pub old_messages_failed_adding_to_backup: u64,
     /// Current retry attempt count
     pub cur_retry_count: u64,
     /// Depth of the ingestion channel
@@ -382,6 +384,7 @@ pub struct SiftStreamMetrics {
     pub(crate) bytes_sent: U64Counter,
     pub(crate) messages_sent_to_backup: U64Counter,
     pub(crate) old_messages_dropped_for_ingestion: U64Counter,
+    pub(crate) old_messages_failed_adding_to_backup: U64Counter,
     pub(crate) cur_retry_count: U64Signal,
     pub(crate) ingestion_channel_depth: U64Signal,
     pub(crate) backup_channel_depth: U64Signal,
@@ -407,6 +410,7 @@ impl SiftStreamMetrics {
         let bytes_sent = self.bytes_sent.get();
         let messages_sent_to_backup = self.messages_sent_to_backup.get();
         let old_messages_dropped_for_ingestion = self.old_messages_dropped_for_ingestion.get();
+        let old_messages_failed_adding_to_backup = self.old_messages_failed_adding_to_backup.get();
         let cur_retry_count = self.cur_retry_count.get();
         let ingestion_channel_depth = self.ingestion_channel_depth.get();
         let backup_channel_depth = self.backup_channel_depth.get();
@@ -424,6 +428,7 @@ impl SiftStreamMetrics {
             byte_rate: stats.byte_rate,
             messages_sent_to_backup,
             old_messages_dropped_for_ingestion,
+            old_messages_failed_adding_to_backup,
             cur_retry_count,
             ingestion_channel_depth,
             backup_channel_depth,
@@ -497,6 +502,12 @@ impl SiftStreamMetricsSnapshot {
                 data_type: ChannelDataType::Uint64.into(),
                 ..Default::default()
             },
+            ChannelConfig {
+                name: format!("{channel_prefix}.old_messages_failed_adding_to_backup"),
+                description: "Old messages failed to add to backup".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
             ChannelConfig {
                 name: format!("{channel_prefix}.cur_retry_count"),
                 description: "Current retry count".into(),
@@ -659,6 +670,10 @@ impl SiftStreamMetricsSnapshot {
                 &format!("{channel_prefix}.old_messages_dropped_for_ingestion"),
                 self.old_messages_dropped_for_ingestion,
             ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.old_messages_failed_adding_to_backup"),
+                self.old_messages_failed_adding_to_backup,
+            ),
             ChannelValue::new(
                 &format!("{channel_prefix}.cur_retry_count"),
                 self.cur_retry_count,
@@ -762,6 +777,7 @@ impl Default for SiftStreamMetrics {
             bytes_sent: U64Counter::default(),
             messages_sent_to_backup: U64Counter::default(),
             old_messages_dropped_for_ingestion: U64Counter::default(),
+            old_messages_failed_adding_to_backup: U64Counter::default(),
             cur_retry_count: U64Signal::default(),
             ingestion_channel_depth: U64Signal::default(),
             backup_channel_depth: U64Signal::default(),

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

@@ -307,12 +307,30 @@ impl SiftStream<IngestionConfigMode> {
 
                 // Re-send the oldest message to the backup to ensure it is re-ingested later despite being
                 // dropped from the ingestion channel.
-                self.mode
+                //
+                // On failure, rely on metrics to track occurences. Logging can quickly become spammy as
+                // the system works through bursts of messages so logs are reduced to the debug level.
+                if let Err(e) = self
+                    .mode
                     .stream_system
                     .backup_tx
                     .try_send(oldest_message)
                     .map_err(|e| Error::new(ErrorKind::StreamError, e))
                     .context("failed to send data to backup task system")
+                {
+                    self.metrics
+                        .old_messages_failed_adding_to_backup
+                        .increment();
+
+                    #[cfg(feature = "tracing")]
+                    tracing::debug!(
+                        sift_stream_id = self.mode.sift_stream_id.to_string(),
+                        "failed to send oldest data to backup task system: {e}"
+                    );
+                }
+
+                // Do not interupt ingestion.
+                Ok(())
             }
             Err(e) => Err(Error::new_msg(
                 ErrorKind::StreamError,
@@ -346,34 +364,74 @@ impl SiftStream<IngestionConfigMode> {
             "adding new flows to ingestion config"
         );
 
-        new_ingestion_config_service(self.grpc_channel.clone())
-            .try_create_flows(
-                &self.mode.ingestion_config.ingestion_config_id,
-                filtered
-                    .iter()
-                    .map(|f| (*f).clone())
-                    .collect::<Vec<FlowConfig>>(),
-            )
-            .await
-            .context("SiftStream::add_new_flows")?;
-
-        self.metrics.loaded_flows.add(filtered.len() as u64);
-
-        for flow_config in filtered {
-            let flow_name = flow_config.name.clone();
+        let mut calls = Vec::with_capacity(filtered.len());
+        let create_flows = filtered.into_iter().cloned().collect::<Vec<FlowConfig>>();
+        for flow_config in create_flows.iter() {
+            let channel = self.grpc_channel.clone();
+            let ingestion_config_id = self.mode.ingestion_config.ingestion_config_id.clone();
+            let flow_config = flow_config.clone();
+
+            calls.push(tokio::spawn(async move {
+                new_ingestion_config_service(channel)
+                    .try_create_flows(&ingestion_config_id, vec![flow_config])
+                    .await
+                    .context("SiftStream::add_new_flows")
+            }));
+        }
+
+        // Wait for all the gRPC calls to complete.
+        let results = futures::future::join_all(calls).await;
+
+        let mut add_config = |config: &FlowConfig| -> Result<()> {
+            let flow_name = config.name.clone();
             let flow_descriptor = FlowDescriptor::try_from((
                 self.mode.ingestion_config.ingestion_config_id.clone(),
-                flow_config,
+                config,
             ))?;
             self.mode.flows_by_name.insert(flow_name, flow_descriptor);
 
             #[cfg(feature = "tracing")]
             tracing::info!(
                 sift_stream_id = self.mode.sift_stream_id.to_string(),
-                flow = flow_config.name,
+                flow = config.name,
                 "successfully registered new flow"
             );
+
+            Ok(())
+        };
+
+        // Iterate over the results and update the flow cache for the successfully created flows.
+        for (config, result) in create_flows.iter().zip(results.into_iter()) {
+            match result {
+                Ok(Ok(())) => {
+                    add_config(config)?;
+                }
+                Ok(Err(e)) if e.kind() == ErrorKind::AlreadyExistsError => {
+                    add_config(config)?;
+                }
+                Ok(Err(e)) => {
+                    #[cfg(feature = "tracing")]
+                    tracing::error!(
+                        sift_stream_id = self.mode.sift_stream_id.to_string(),
+                        "failed to create flow {}: {e}",
+                        config.name,
+                    );
+                }
+                Err(e) => {
+                    #[cfg(feature = "tracing")]
+                    tracing::error!(
+                        sift_stream_id = self.mode.sift_stream_id.to_string(),
+                        "failed to create flow {}: {e}",
+                        config.name,
+                    );
+                }
+            }
         }
+
+        self.metrics
+            .loaded_flows
+            .add(self.mode.flows_by_name.len() as u64);
+
         Ok(())
     }
 

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

@@ -4,7 +4,8 @@ use std::time::Duration;
 use crate::TimeValue;
 use crate::backup::DiskBackupPolicy;
 use crate::{
-    ChannelValue, Flow, IngestionConfigForm, RecoveryStrategy, RunForm, SiftStreamBuilder,
+    ChannelValue, Flow, FlowBuilder, IngestionConfigForm, RecoveryStrategy, RunForm,
+    SiftStreamBuilder,
 };
 use sift_rs::common::r#type::v1::ChannelDataType;
 use sift_rs::ingestion_configs::v2::{ChannelConfig, FlowConfig};
@@ -399,3 +400,78 @@ async fn test_sift_stream_builder_load_ingestion_config_with_new_flows() {
     let flows = sift_stream.get_flows();
     assert_eq!(flows.len(), 2);
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn test_sift_stream_ingestion_and_backup_channels_fill_up() {
+    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 existing_flow = FlowConfig {
+        name: "already_exists_flow".to_string(),
+        channels: vec![ChannelConfig {
+            name: "channel1".to_string(),
+            data_type: ChannelDataType::Double.into(),
+            ..Default::default()
+        }],
+    };
+
+    let ingestion_config = IngestionConfigForm {
+        asset_name: "test_asset".to_string(),
+        client_key: "test_client_key".to_string(),
+        flows: vec![existing_flow],
+    };
+    let disk_backup_policy = DiskBackupPolicy {
+        backups_dir: Some(tmp_dir_path.to_path_buf()),
+        retain_backups: true,
+        ..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,
+        })
+        .metrics_streaming_interval(None)
+        .ingestion_data_channel_capacity(1)
+        .backup_data_channel_capacity(1)
+        .build()
+        .await
+        .expect("failed to build sift stream");
+
+    let descriptor = sift_stream
+        .get_flow_descriptor("already_exists_flow")
+        .expect("failed to get flow descriptor");
+
+    // Send a burst of messages that will cause the ingestion and backup channels to fill up.
+    //
+    // Since this test is running in single-threded mode, and `send_requests_nonblocking` is not async,
+    // sending all the messages should occur before the background tasks have a chance to run
+    // and create space.
+    for data in 0..100 {
+        let mut builder = FlowBuilder::new(&descriptor);
+        assert!(builder.set_with_key("channel1", data as f64).is_ok());
+
+        assert!(
+            sift_stream
+                .send_requests(vec![builder.request(TimeValue::now())])
+                .await
+                .is_ok(),
+            "failed to send request"
+        );
+    }
+
+    // Finish the stream to ensure that the backup manager is shutdown and the backup files are processed.
+    tokio::time::timeout(Duration::from_secs(10), async {
+        assert!(
+            sift_stream.finish().await.is_ok(),
+            "failed to finish sift stream"
+        );
+    })
+    .await
+    .expect("timeout waiting for sift stream to finish");
+}