fix: Initialize rustls's CryptoProvider early in the code (#2312)

Signed-off-by: Sreekanth <[email protected]>

Author: BulkBeing

rust/Cargo.lock

@@ -6,7 +6,8 @@ edition = "2021"
 [features]
 nats-tests = []
 pulsar-tests = []
-all-tests = ["nats-tests", "pulsar-tests"]
+redis-tests = []
+all-tests = ["nats-tests", "pulsar-tests", "redis-tests"]
 
 [lints]
 workspace = true

rust/numaflow-core/Cargo.toml

@@ -139,13 +139,18 @@ mod tests {
         }
     }
 
+    #[cfg(feature = "redis-tests")]
     #[tokio::test]
     async fn test_serving_source_reader_acker() -> Result<()> {
         let settings = Settings {
             app_listen_port: 2000,
             ..Default::default()
         };
         let settings = Arc::new(settings);
+        // Setup the CryptoProvider (controls core cryptography used by rustls) for the process
+        // ServingSource starts an Axum HTTPS server in the background. Rustls is used to generate
+        // self-signed certs when starting the server.
+        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
         let mut serving_source = ServingSource::new(
             Arc::clone(&settings),
             10,

rust/numaflow-core/src/source/serving.rs

@@ -14,4 +14,5 @@ numaflow-models.workspace = true
 backoff.workspace = true
 tokio.workspace = true
 tracing.workspace = true
+rustls.workspace = true
 tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }

rust/numaflow/Cargo.toml

@@ -19,6 +19,12 @@ async fn main() -> Result<(), Box<dyn Error>> {
         )
         .with(tracing_subscriber::fmt::layer().with_ansi(false))
         .init();
+
+    // Setup the CryptoProvider (controls core cryptography used by rustls) for the process
+    rustls::crypto::aws_lc_rs::default_provider()
+        .install_default()
+        .expect("Installing default CryptoProvider");
+
     if let Err(e) = run().await {
         error!("{e:?}");
         return Err(e);

rust/numaflow/src/main.rs

@@ -32,15 +32,13 @@ use crate::{app::callback::state, Message, MessageWrapper};
 //   "from_vertex": "a"
 // }
 
-const CALLBACK_URL_KEY: &str = "X-Numaflow-Callback-Url";
 const NUMAFLOW_RESP_ARRAY_LEN: &str = "Numaflow-Array-Len";
 const NUMAFLOW_RESP_ARRAY_IDX_LEN: &str = "Numaflow-Array-Index-Len";
 
 struct ProxyState<T> {
     message: mpsc::Sender<MessageWrapper>,
     tid_header: String,
     callback: state::State<T>,
-    callback_url: String,
 }
 
 pub(crate) async fn jetstream_proxy<T: Clone + Send + Sync + Store + 'static>(
@@ -50,10 +48,6 @@ pub(crate) async fn jetstream_proxy<T: Clone + Send + Sync + Store + 'static>(
         message: state.message.clone(),
         tid_header: state.settings.tid_header.clone(),
         callback: state.callback_state.clone(),
-        callback_url: format!(
-            "https://{}:{}/v1/process/callback",
-            state.settings.host_ip, state.settings.app_listen_port
-        ),
     });
 
     let router = Router::new()

rust/serving/src/app/jetstream_proxy.rs

@@ -39,9 +39,6 @@ pub(crate) async fn serve<T>(
 where
     T: Clone + Send + Sync + Store + 'static,
 {
-    // Setup the CryptoProvider (controls core cryptography used by rustls) for the process
-    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
-
     let (cert, key) = generate_certs()?;
 
     let tls_config = RustlsConfig::from_pem(cert.pem().into(), key.serialize_pem().into())

rust/serving/src/lib.rs

@@ -65,7 +65,7 @@ pub(crate) struct Edge {
 /// DCG (directed compute graph) of the pipeline with minimal information build using vertices and edges
 /// from the pipeline spec
 #[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq)]
-pub(crate) struct PipelineDCG {
+pub struct PipelineDCG {
     pub(crate) vertices: Vec<Vertex>,
     pub(crate) edges: Vec<Edge>,
 }