feat: change async-bb8 fork and tokio spawn for concurrent database calls (#2774)

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: akshay-97 <[email protected]>
Co-authored-by: akshay.s <[email protected]>
Co-authored-by: Kartikeya Hegde <[email protected]>

Author: 5 people

Cargo.lock

@@ -77,7 +77,9 @@ impl<const PRECISION: u8> Percentage<PRECISION> {
         if value.contains('.') {
             // if string has '.' then take the decimal part and verify precision length
             match value.split('.').last() {
-                Some(decimal_part) => decimal_part.trim_end_matches('0').len() <= PRECISION.into(),
+                Some(decimal_part) => {
+                    decimal_part.trim_end_matches('0').len() <= <u8 as Into<usize>>::into(PRECISION)
+                }
                 // will never be None
                 None => false,
             }

crates/common_utils/src/types.rs

@@ -12,7 +12,7 @@ default = ["kv_store"]
 kv_store = []
 
 [dependencies]
-async-bb8-diesel = "0.1.0"
+async-bb8-diesel = { git = "https://github.com/jarnura/async-bb8-diesel", rev = "53b4ab901aab7635c8215fd1c2d542c8db443094" }
 diesel = { version = "2.1.0", features = ["postgres", "serde_json", "time", "64-column-tables"] }
 error-stack = "0.3.1"
 frunk = "0.4.1"

crates/diesel_models/Cargo.toml

@@ -13,7 +13,7 @@ kms = ["external_services/kms"]
 vergen = ["router_env/vergen"]
 
 [dependencies]
-async-bb8-diesel = "0.1.0"
+async-bb8-diesel = { git = "https://github.com/jarnura/async-bb8-diesel", rev = "53b4ab901aab7635c8215fd1c2d542c8db443094" }
 bb8 = "0.8"
 clap = { version = "4.3.2", default-features = false, features = ["std", "derive", "help", "usage"] }
 config = { version = "0.13.3", features = ["toml"] }

crates/drainer/Cargo.toml

@@ -37,8 +37,8 @@ actix-cors = "0.6.4"
 actix-multipart = "0.6.0"
 actix-rt = "2.8.0"
 actix-web = "4.3.1"
+async-bb8-diesel = { git = "https://github.com/jarnura/async-bb8-diesel", rev = "53b4ab901aab7635c8215fd1c2d542c8db443094" }
 argon2 = { version = "0.5.0", features = ["std"] }
-async-bb8-diesel = "0.1.0"
 async-trait = "0.1.68"
 aws-config = { version = "0.55.3", optional = true }
 aws-sdk-s3 = { version = "0.28.0", optional = true }
@@ -97,6 +97,7 @@ utoipa-swagger-ui = { version = "3.1.3", features = ["actix-web"] }
 uuid = { version = "1.3.3", features = ["serde", "v4"] }
 validator = "0.16.0"
 x509-parser = "0.15.0"
+tracing-futures = { version = "0.2.5", features = ["tokio"] }
 
 # First party crates
 api_models = { version = "0.1.0", path = "../api_models", features = ["errors"] }

crates/router/Cargo.toml

@@ -4,7 +4,7 @@ use router::{
     logger,
 };
 
-#[actix_web::main]
+#[tokio::main]
 async fn main() -> ApplicationResult<()> {
     // get commandline config before initializing config
     let cmd_line = <CmdLineConf as clap::Parser>::parse();
@@ -43,7 +43,7 @@ async fn main() -> ApplicationResult<()> {
     logger::info!("Application started [{:?}] [{:?}]", conf.server, conf.log);
 
     #[allow(clippy::expect_used)]
-    let server = router::start_server(conf)
+    let server = Box::pin(router::start_server(conf))
         .await
         .expect("Failed to create the server");
     let _ = server.await;

crates/router/src/bin/router.rs

@@ -40,7 +40,12 @@ async fn main() -> CustomResult<(), ProcessTrackerError> {
     );
     // channel for listening to redis disconnect events
     let (redis_shutdown_signal_tx, redis_shutdown_signal_rx) = oneshot::channel();
-    let state = routes::AppState::new(conf, redis_shutdown_signal_tx, api_client).await;
+    let state = Box::pin(routes::AppState::new(
+        conf,
+        redis_shutdown_signal_tx,
+        api_client,
+    ))
+    .await;
     // channel to shutdown scheduler gracefully
     let (tx, rx) = mpsc::channel(1);
     tokio::spawn(router::receiver_for_error(

crates/router/src/bin/scheduler.rs

@@ -1944,7 +1944,14 @@ pub async fn do_list_customer_pm_fetch_customer_if_not_passed(
 ) -> errors::RouterResponse<api::CustomerPaymentMethodsListResponse> {
     let db = state.store.as_ref();
     if let Some(customer_id) = customer_id {
-        list_customer_payment_method(&state, merchant_account, key_store, None, customer_id).await
+        Box::pin(list_customer_payment_method(
+            &state,
+            merchant_account,
+            key_store,
+            None,
+            customer_id,
+        ))
+        .await
     } else {
         let cloned_secret = req.and_then(|r| r.client_secret.as_ref().cloned());
         let payment_intent = helpers::verify_payment_intent_time_and_client_secret(
@@ -1957,13 +1964,13 @@ pub async fn do_list_customer_pm_fetch_customer_if_not_passed(
             .as_ref()
             .and_then(|intent| intent.customer_id.to_owned())
             .ok_or(errors::ApiErrorResponse::CustomerNotFound)?;
-        list_customer_payment_method(
+        Box::pin(list_customer_payment_method(
             &state,
             merchant_account,
             key_store,
             payment_intent,
             &customer_id,
-        )
+        ))
         .await
     }
 }

crates/router/src/core/payment_methods/cards.rs

@@ -193,15 +193,15 @@ where
                 )
                 .await?;
                 let operation = Box::new(PaymentResponse);
-                let db = &*state.store;
+
                 connector_http_status_code = router_data.connector_http_status_code;
                 external_latency = router_data.external_latency;
                 //add connector http status code metrics
                 add_connector_http_status_code_metrics(connector_http_status_code);
                 operation
                     .to_post_update_tracker()?
                     .update_tracker(
-                        db,
+                        state,
                         &validate_result.payment_id,
                         payment_data,
                         router_data,
@@ -272,15 +272,14 @@ where
                 }
 
                 let operation = Box::new(PaymentResponse);
-                let db = &*state.store;
                 connector_http_status_code = router_data.connector_http_status_code;
                 external_latency = router_data.external_latency;
                 //add connector http status code metrics
                 add_connector_http_status_code_metrics(connector_http_status_code);
                 operation
                     .to_post_update_tracker()?
                     .update_tracker(
-                        db,
+                        state,
                         &validate_result.payment_id,
                         payment_data,
                         router_data,
@@ -323,7 +322,7 @@ where
         (_, payment_data) = operation
             .to_update_tracker()?
             .update_trackers(
-                &*state.store,
+                state,
                 payment_data.clone(),
                 customer.clone(),
                 validate_result.storage_scheme,
@@ -582,7 +581,14 @@ impl<Ctx: PaymentMethodRetrieve> PaymentRedirectFlow<Ctx> for PaymentRedirectCom
             }),
             ..Default::default()
         };
-        payments_core::<api::CompleteAuthorize, api::PaymentsResponse, _, _, _, Ctx>(
+        Box::pin(payments_core::<
+            api::CompleteAuthorize,
+            api::PaymentsResponse,
+            _,
+            _,
+            _,
+            Ctx,
+        >(
             state.clone(),
             merchant_account,
             merchant_key_store,
@@ -592,7 +598,7 @@ impl<Ctx: PaymentMethodRetrieve> PaymentRedirectFlow<Ctx> for PaymentRedirectCom
             connector_action,
             None,
             HeaderPayload::default(),
-        )
+        ))
         .await
     }
 
@@ -678,7 +684,14 @@ impl<Ctx: PaymentMethodRetrieve> PaymentRedirectFlow<Ctx> for PaymentRedirectSyn
             expand_attempts: None,
             expand_captures: None,
         };
-        payments_core::<api::PSync, api::PaymentsResponse, _, _, _, Ctx>(
+        Box::pin(payments_core::<
+            api::PSync,
+            api::PaymentsResponse,
+            _,
+            _,
+            _,
+            Ctx,
+        >(
             state.clone(),
             merchant_account,
             merchant_key_store,
@@ -688,7 +701,7 @@ impl<Ctx: PaymentMethodRetrieve> PaymentRedirectFlow<Ctx> for PaymentRedirectSyn
             connector_action,
             None,
             HeaderPayload::default(),
-        )
+        ))
         .await
     }
     fn generate_response(
@@ -889,7 +902,7 @@ where
     (_, *payment_data) = operation
         .to_update_tracker()?
         .update_trackers(
-            &*state.store,
+            state,
             payment_data.clone(),
             customer.clone(),
             merchant_account.storage_scheme,

crates/router/src/core/payments.rs

@@ -25,15 +25,18 @@ impl
         customer: &Option<domain::Customer>,
         merchant_connector_account: &helpers::MerchantConnectorAccountType,
     ) -> RouterResult<types::PaymentsApproveRouterData> {
-        transformers::construct_payment_router_data::<api::Approve, types::PaymentsApproveData>(
+        Box::pin(transformers::construct_payment_router_data::<
+            api::Approve,
+            types::PaymentsApproveData,
+        >(
             state,
             self.clone(),
             connector_id,
             merchant_account,
             key_store,
             customer,
             merchant_connector_account,
-        )
+        ))
         .await
     }
 }