feat: change async-bb8 fork and tokio spawn for concurrent database calls

crates/common_utils/src/types.rs

@@ -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/diesel_models/Cargo.toml

@@ -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/drainer/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/router/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/src/bin/router.rs

@@ -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/scheduler.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/core/payment_methods/cards.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/payments.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/flows/approve_flow.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
     }
 }

crates/router/src/core/payments/flows/authorize_flow.rs

@@ -39,15 +39,18 @@ impl
             types::PaymentsResponseData,
         >,
     > {
-        transformers::construct_payment_router_data::<api::Authorize, types::PaymentsAuthorizeData>(
+        Box::pin(transformers::construct_payment_router_data::<
+            api::Authorize,
+            types::PaymentsAuthorizeData,
+        >(
             state,
             self.clone(),
             connector_id,
             merchant_account,
             key_store,
             customer,
             merchant_connector_account,
-        )
+        ))
         .await
     }
 }
@@ -96,15 +99,15 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
             metrics::PAYMENT_COUNT.add(&metrics::CONTEXT, 1, &[]); // Metrics
 
             if resp.request.setup_mandate_details.clone().is_some() {
-                let payment_method_id = tokenization::save_payment_method(
+                let payment_method_id = Box::pin(tokenization::save_payment_method(
                     state,
                     connector,
                     resp.to_owned(),
                     maybe_customer,
                     merchant_account,
                     self.request.payment_method_type,
                     key_store,
-                )
+                ))
                 .await?;
                 Ok(mandate::mandate_procedure(
                     state,
@@ -127,15 +130,15 @@ impl Feature<api::Authorize, types::PaymentsAuthorizeData> for types::PaymentsAu
                 tokio::spawn(async move {
                     logger::info!("Starting async call to save_payment_method in locker");
 
-                    let result = tokenization::save_payment_method(
+                    let result = Box::pin(tokenization::save_payment_method(
                         &state,
                         &connector,
                         response,
                         &maybe_customer,
                         &merchant_account,
                         self.request.payment_method_type,
                         &key_store,
-                    )
+                    ))
                     .await;
 
                     if let Err(err) = result {

crates/router/src/core/payments/flows/cancel_flow.rs

@@ -24,15 +24,18 @@ impl ConstructFlowSpecificData<api::Void, types::PaymentsCancelData, types::Paym
         customer: &Option<domain::Customer>,
         merchant_connector_account: &helpers::MerchantConnectorAccountType,
     ) -> RouterResult<types::PaymentsCancelRouterData> {
-        transformers::construct_payment_router_data::<api::Void, types::PaymentsCancelData>(
+        Box::pin(transformers::construct_payment_router_data::<
+            api::Void,
+            types::PaymentsCancelData,
+        >(
             state,
             self.clone(),
             connector_id,
             merchant_account,
             key_store,
             customer,
             merchant_connector_account,
-        )
+        ))
         .await
     }
 }

crates/router/src/core/payments/flows/capture_flow.rs

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

crates/router/src/core/payments/flows/complete_authorize_flow.rs

@@ -35,7 +35,7 @@ impl
             types::PaymentsResponseData,
         >,
     > {
-        transformers::construct_payment_router_data::<
+        Box::pin(transformers::construct_payment_router_data::<
             api::CompleteAuthorize,
             types::CompleteAuthorizeData,
         >(
@@ -46,7 +46,7 @@ impl
             key_store,
             customer,
             merchant_connector_account,
-        )
+        ))
         .await
     }
 }

crates/router/src/core/payments/flows/psync_flow.rs

@@ -28,15 +28,18 @@ impl ConstructFlowSpecificData<api::PSync, types::PaymentsSyncData, types::Payme
     ) -> RouterResult<
         types::RouterData<api::PSync, types::PaymentsSyncData, types::PaymentsResponseData>,
     > {
-        transformers::construct_payment_router_data::<api::PSync, types::PaymentsSyncData>(
+        Box::pin(transformers::construct_payment_router_data::<
+            api::PSync,
+            types::PaymentsSyncData,
+        >(
             state,
             self.clone(),
             connector_id,
             merchant_account,
             key_store,
             customer,
             merchant_connector_account,
-        )
+        ))
         .await
     }
 }

crates/router/src/core/payments/flows/reject_flow.rs

@@ -24,15 +24,18 @@ impl ConstructFlowSpecificData<api::Reject, types::PaymentsRejectData, types::Pa
         customer: &Option<domain::Customer>,
         merchant_connector_account: &helpers::MerchantConnectorAccountType,
     ) -> RouterResult<types::PaymentsRejectRouterData> {
-        transformers::construct_payment_router_data::<api::Reject, types::PaymentsRejectData>(
+        Box::pin(transformers::construct_payment_router_data::<
+            api::Reject,
+            types::PaymentsRejectData,
+        >(
             state,
             self.clone(),
             connector_id,
             merchant_account,
             key_store,
             customer,
             merchant_connector_account,
-        )
+        ))
         .await
     }
 }