feat(dynamic_routing): integration of elimination routing for core flows (#6816)

Co-authored-by: Aprabhat19 <amishaprabhat@gmail.com>
Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>
Co-authored-by: Amisha Prabhat <55580080+Aprabhat19@users.noreply.github.com>
Co-authored-by: Chethan Rao <70657455+Chethan-rao@users.noreply.github.com>

config/development.toml

@@ -1016,7 +1016,7 @@ connector_list = "cybersource"
 
 [grpc_client.dynamic_routing_client]
 host = "localhost"
-port = 7000
+port = 8000
 service = "dynamo"
 
 [theme.storage]

crates/api_models/src/routing.rs

@@ -811,7 +811,7 @@ pub struct EliminationRoutingConfig {
     pub elimination_analyser_config: Option<EliminationAnalyserConfig>,
 }
 
-#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, ToSchema)]
+#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, Copy, ToSchema)]
 pub struct EliminationAnalyserConfig {
     pub bucket_size: Option<u64>,
     pub bucket_leak_interval_in_secs: Option<u64>,

crates/common_enums/src/enums.rs

@@ -7795,6 +7795,17 @@ pub enum ErrorCategory {
     ProcessorDeclineIncorrectData,
 }
 
+impl ErrorCategory {
+    pub fn should_perform_elimination_routing(self) -> bool {
+        match self {
+            Self::ProcessorDowntime | Self::ProcessorDeclineUnauthorized => true,
+            Self::IssueWithPaymentMethod
+            | Self::ProcessorDeclineIncorrectData
+            | Self::FrmDecline => false,
+        }
+    }
+}
+
 #[derive(
     Clone,
     Debug,

crates/router/src/core/errors.rs

@@ -387,6 +387,18 @@ pub enum RoutingError {
     SuccessRateCalculationError,
     #[error("Success rate client from dynamic routing gRPC service not initialized")]
     SuccessRateClientInitializationError,
+    #[error("Elimination client from dynamic routing gRPC service not initialized")]
+    EliminationClientInitializationError,
+    #[error("Unable to analyze elimination routing config from dynamic routing service")]
+    EliminationRoutingCalculationError,
+    #[error("Params not found in elimination based routing config")]
+    EliminationBasedRoutingParamsNotFoundError,
+    #[error("Unable to retrieve elimination based routing config")]
+    EliminationRoutingConfigError,
+    #[error(
+        "Invalid elimination based connector label received from dynamic routing service: '{0}'"
+    )]
+    InvalidEliminationBasedConnectorLabel(String),
     #[error("Unable to convert from '{from}' to '{to}'")]
     GenericConversionError { from: String, to: String },
     #[error("Invalid success based connector label received from dynamic routing service: '{0}'")]

crates/router/src/core/payments/operations/payment_response.rs

@@ -1415,7 +1415,13 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
             .as_mut()
             .map(|info| info.status = status)
     });
-    let (capture_update, mut payment_attempt_update) = match router_data.response.clone() {
+
+    // TODO: refactor of gsm_error_category with respective feature flag
+    #[allow(unused_variables)]
+    let (capture_update, mut payment_attempt_update, gsm_error_category) = match router_data
+        .response
+        .clone()
+    {
         Err(err) => {
             let auth_update = if Some(router_data.auth_type)
                 != payment_data.payment_attempt.authentication_type
@@ -1424,7 +1430,8 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
             } else {
                 None
             };
-            let (capture_update, attempt_update) = match payment_data.multiple_capture_data {
+            let (capture_update, attempt_update, gsm_error_category) =
+                match payment_data.multiple_capture_data {
                     Some(multiple_capture_data) => {
                         let capture_update = storage::CaptureUpdate::ErrorUpdate {
                             status: match err.status_code {
@@ -1447,6 +1454,7 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                                     updated_by: storage_scheme.to_string(),
                                 }
                             }),
+                            None,
                         )
                     }
                     None => {
@@ -1463,7 +1471,8 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
 
                         let gsm_unified_code =
                             option_gsm.as_ref().and_then(|gsm| gsm.unified_code.clone());
-                    let gsm_unified_message = option_gsm.and_then(|gsm| gsm.unified_message);
+                        let gsm_unified_message =
+                            option_gsm.clone().and_then(|gsm| gsm.unified_message);
 
                         let (unified_code, unified_message) = if let Some((code, message)) =
                             gsm_unified_code.as_ref().zip(gsm_unified_message.as_ref())
@@ -1537,10 +1546,11 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                                 issuer_error_code: err.network_decline_code,
                                 issuer_error_message: err.network_error_message,
                             }),
+                            option_gsm.and_then(|option_gsm| option_gsm.error_category),
                         )
                     }
                 };
-            (capture_update, attempt_update)
+            (capture_update, attempt_update, gsm_error_category)
         }
 
         Ok(payments_response) => {
@@ -1576,6 +1586,7 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                             issuer_error_code: None,
                             issuer_error_message: None,
                         }),
+                        None,
                     )
                 }
                 Ok(()) => {
@@ -1631,7 +1642,7 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                                     updated_by: storage_scheme.to_string(),
                                 };
 
-                            (None, Some(payment_attempt_update))
+                            (None, Some(payment_attempt_update), None)
                         }
                         types::PaymentsResponseData::TransactionResponse {
                             resource_id,
@@ -1852,7 +1863,7 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                                 ),
                             };
 
-                            (capture_updates, payment_attempt_update)
+                            (capture_updates, payment_attempt_update, None)
                         }
                         types::PaymentsResponseData::TransactionUnresolvedResponse {
                             resource_id,
@@ -1880,23 +1891,30 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                                     connector_response_reference_id,
                                     updated_by: storage_scheme.to_string(),
                                 }),
+                                None,
                             )
                         }
-                        types::PaymentsResponseData::SessionResponse { .. } => (None, None),
+                        types::PaymentsResponseData::SessionResponse { .. } => (None, None, None),
-                        types::PaymentsResponseData::SessionTokenResponse { .. } => (None, None),
+                        types::PaymentsResponseData::SessionTokenResponse { .. } => {
-                        types::PaymentsResponseData::TokenizationResponse { .. } => (None, None),
+                            (None, None, None)
+                        }
+                        types::PaymentsResponseData::TokenizationResponse { .. } => {
+                            (None, None, None)
+                        }
                         types::PaymentsResponseData::ConnectorCustomerResponse { .. } => {
-                            (None, None)
+                            (None, None, None)
                         }
                         types::PaymentsResponseData::ThreeDSEnrollmentResponse { .. } => {
-                            (None, None)
+                            (None, None, None)
+                        }
+                        types::PaymentsResponseData::PostProcessingResponse { .. } => {
+                            (None, None, None)
                         }
-                        types::PaymentsResponseData::PostProcessingResponse { .. } => (None, None),
                         types::PaymentsResponseData::IncrementalAuthorizationResponse {
                             ..
-                        } => (None, None),
+                        } => (None, None, None),
                         types::PaymentsResponseData::PaymentResourceUpdateResponse { .. } => {
-                            (None, None)
+                            (None, None, None)
                         }
                         types::PaymentsResponseData::MultipleCaptureResponse {
                             capture_sync_response_list,
@@ -1906,9 +1924,13 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                                     &multiple_capture_data,
                                     capture_sync_response_list,
                                 )?;
-                                (Some((multiple_capture_data, capture_update_list)), None)
+                                (
+                                    Some((multiple_capture_data, capture_update_list)),
+                                    None,
+                                    None,
+                                )
                             }
-                            None => (None, None),
+                            None => (None, None, None),
                         },
                     }
                 }
@@ -2146,6 +2168,23 @@ async fn payment_response_update_tracker<F: Clone, T: types::Capturable>(
                     .map_err(|e| logger::error!(success_based_routing_metrics_error=?e))
                     .ok();
 
+                    if let Some(gsm_error_category) = gsm_error_category {
+                        if gsm_error_category.should_perform_elimination_routing() {
+                            logger::info!("Performing update window for elimination routing");
+                            routing_helpers::update_window_for_elimination_routing(
+                                &state,
+                                &payment_attempt,
+                                &profile_id,
+                                dynamic_routing_algo_ref.clone(),
+                                dynamic_routing_config_params_interpolator.clone(),
+                                gsm_error_category,
+                            )
+                            .await
+                            .map_err(|e| logger::error!(dynamic_routing_metrics_error=?e))
+                            .ok();
+                        };
+                    };
+
                     routing_helpers::push_metrics_with_update_window_for_contract_based_routing(
                         &state,
                         &payment_attempt,

crates/router/src/core/payments/routing.rs

@@ -32,6 +32,7 @@ use euclid::{
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 use external_services::grpc_client::dynamic_routing::{
     contract_routing_client::ContractBasedDynamicRouting,
+    elimination_based_client::{EliminationBasedRouting, EliminationResponse},
     success_rate_client::{CalSuccessRateResponse, SuccessBasedDynamicRouting},
     DynamicRoutingError,
 };
@@ -1562,7 +1563,7 @@ pub async fn perform_dynamic_routing(
         profile.get_id().get_string_repr()
     );
 
-    let connector_list = match dynamic_routing_algo_ref
+    let mut connector_list = match dynamic_routing_algo_ref
         .success_based_algorithm
         .as_ref()
         .async_map(|algorithm| {
@@ -1591,7 +1592,7 @@ pub async fn perform_dynamic_routing(
                         state,
                         routable_connectors.clone(),
                         profile.get_id(),
-                        dynamic_routing_config_params_interpolator,
+                        dynamic_routing_config_params_interpolator.clone(),
                         algorithm.clone(),
                     )
                 })
@@ -1600,10 +1601,29 @@ pub async fn perform_dynamic_routing(
                 .inspect_err(|e| logger::error!(dynamic_routing_error=?e))
                 .ok()
                 .flatten()
-                .unwrap_or(routable_connectors)
+                .unwrap_or(routable_connectors.clone())
         }
     };
 
+    connector_list = dynamic_routing_algo_ref
+        .elimination_routing_algorithm
+        .as_ref()
+        .async_map(|algorithm| {
+            perform_elimination_routing(
+                state,
+                connector_list.clone(),
+                profile.get_id(),
+                dynamic_routing_config_params_interpolator.clone(),
+                algorithm.clone(),
+            )
+        })
+        .await
+        .transpose()
+        .inspect_err(|e| logger::error!(dynamic_routing_error=?e))
+        .ok()
+        .flatten()
+        .unwrap_or(connector_list);
+
     Ok(connector_list)
 }
 
@@ -1865,6 +1885,127 @@ pub async fn perform_success_based_routing(
     }
 }
 
+/// elimination dynamic routing
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
+pub async fn perform_elimination_routing(
+    state: &SessionState,
+    routable_connectors: Vec<api_routing::RoutableConnectorChoice>,
+    profile_id: &common_utils::id_type::ProfileId,
+    elimination_routing_configs_params_interpolator: routing::helpers::DynamicRoutingConfigParamsInterpolator,
+    elimination_algo_ref: api_routing::EliminationRoutingAlgorithm,
+) -> RoutingResult<Vec<api_routing::RoutableConnectorChoice>> {
+    if elimination_algo_ref.enabled_feature
+        == api_routing::DynamicRoutingFeatures::DynamicConnectorSelection
+    {
+        logger::debug!(
+            "performing elimination_routing for profile {}",
+            profile_id.get_string_repr()
+        );
+        let client = state
+            .grpc_client
+            .dynamic_routing
+            .elimination_based_client
+            .as_ref()
+            .ok_or(errors::RoutingError::EliminationClientInitializationError)
+            .attach_printable("elimination routing's gRPC client not found")?;
+
+        let elimination_routing_config = routing::helpers::fetch_dynamic_routing_configs::<
+            api_routing::EliminationRoutingConfig,
+        >(
+            state,
+            profile_id,
+            elimination_algo_ref
+                .algorithm_id_with_timestamp
+                .algorithm_id
+                .ok_or(errors::RoutingError::GenericNotFoundError {
+                    field: "elimination_routing_algorithm_id".to_string(),
+                })
+                .attach_printable(
+                    "elimination_routing_algorithm_id not found in business_profile",
+                )?,
+        )
+        .await
+        .change_context(errors::RoutingError::EliminationRoutingConfigError)
+        .attach_printable("unable to fetch elimination dynamic routing configs")?;
+
+        let elimination_routing_config_params = elimination_routing_configs_params_interpolator
+            .get_string_val(
+                elimination_routing_config
+                    .params
+                    .as_ref()
+                    .ok_or(errors::RoutingError::EliminationBasedRoutingParamsNotFoundError)?,
+            );
+
+        let elimination_based_connectors: EliminationResponse = client
+            .perform_elimination_routing(
+                profile_id.get_string_repr().to_string(),
+                elimination_routing_config_params,
+                routable_connectors.clone(),
+                elimination_routing_config.elimination_analyser_config,
+                state.get_grpc_headers(),
+            )
+            .await
+            .change_context(errors::RoutingError::EliminationRoutingCalculationError)
+            .attach_printable(
+                "unable to analyze/fetch elimination routing from dynamic routing service",
+            )?;
+        let mut connectors =
+            Vec::with_capacity(elimination_based_connectors.labels_with_status.len());
+        let mut eliminated_connectors =
+            Vec::with_capacity(elimination_based_connectors.labels_with_status.len());
+        let mut non_eliminated_connectors =
+            Vec::with_capacity(elimination_based_connectors.labels_with_status.len());
+        for labels_with_status in elimination_based_connectors.labels_with_status {
+            let (connector, merchant_connector_id) = labels_with_status.label
+                .split_once(':')
+                .ok_or(errors::RoutingError::InvalidEliminationBasedConnectorLabel(labels_with_status.label.to_string()))
+                .attach_printable(
+                    "unable to split connector_name and mca_id from the label obtained by the elimination based dynamic routing service",
+                )?;
+
+            let routable_connector = api_routing::RoutableConnectorChoice {
+                choice_kind: api_routing::RoutableChoiceKind::FullStruct,
+                connector: common_enums::RoutableConnectors::from_str(connector)
+                    .change_context(errors::RoutingError::GenericConversionError {
+                        from: "String".to_string(),
+                        to: "RoutableConnectors".to_string(),
+                    })
+                    .attach_printable("unable to convert String to RoutableConnectors")?,
+                merchant_connector_id: Some(
+                    common_utils::id_type::MerchantConnectorAccountId::wrap(
+                        merchant_connector_id.to_string(),
+                    )
+                    .change_context(errors::RoutingError::GenericConversionError {
+                        from: "String".to_string(),
+                        to: "MerchantConnectorAccountId".to_string(),
+                    })
+                    .attach_printable("unable to convert MerchantConnectorAccountId from string")?,
+                ),
+            };
+
+            if labels_with_status
+                .elimination_information
+                .is_some_and(|elimination_info| {
+                    elimination_info
+                        .entity
+                        .is_some_and(|entity_info| entity_info.is_eliminated)
+                })
+            {
+                eliminated_connectors.push(routable_connector);
+            } else {
+                non_eliminated_connectors.push(routable_connector);
+            }
+            connectors.extend(non_eliminated_connectors.clone());
+            connectors.extend(eliminated_connectors.clone());
+        }
+        logger::debug!(dynamic_eliminated_connectors=?eliminated_connectors);
+        logger::debug!(dynamic_elimination_based_routing_connectors=?connectors);
+        Ok(connectors)
+    } else {
+        Ok(routable_connectors)
+    }
+}
+
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 pub async fn perform_contract_based_routing(
     state: &SessionState,

crates/router/src/core/routing/helpers.rs

@@ -21,9 +21,10 @@ use error_stack::ResultExt;
 #[cfg(all(feature = "dynamic_routing", feature = "v1"))]
 use external_services::grpc_client::dynamic_routing::{
     contract_routing_client::ContractBasedDynamicRouting,
+    elimination_based_client::EliminationBasedRouting,
     success_rate_client::SuccessBasedDynamicRouting,
 };
-#[cfg(all(feature = "dynamic_routing", feature = "v1"))]
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 use hyperswitch_domain_models::api::ApplicationResponse;
 #[cfg(all(feature = "dynamic_routing", feature = "v1"))]
 use router_env::logger;
@@ -610,6 +611,43 @@ impl DynamicRoutingCache for routing_types::ContractBasedRoutingConfig {
     }
 }
 
+#[cfg(all(feature = "dynamic_routing", feature = "v1"))]
+#[async_trait::async_trait]
+impl DynamicRoutingCache for routing_types::EliminationRoutingConfig {
+    async fn get_cached_dynamic_routing_config_for_profile(
+        state: &SessionState,
+        key: &str,
+    ) -> Option<Arc<Self>> {
+        cache::ELIMINATION_BASED_DYNAMIC_ALGORITHM_CACHE
+            .get_val::<Arc<Self>>(cache::CacheKey {
+                key: key.to_string(),
+                prefix: state.tenant.redis_key_prefix.clone(),
+            })
+            .await
+    }
+
+    async fn refresh_dynamic_routing_cache<T, F, Fut>(
+        state: &SessionState,
+        key: &str,
+        func: F,
+    ) -> RouterResult<T>
+    where
+        F: FnOnce() -> Fut + Send,
+        T: Cacheable + serde::Serialize + serde::de::DeserializeOwned + Debug + Clone,
+        Fut: futures::Future<Output = errors::CustomResult<T, errors::StorageError>> + Send,
+    {
+        cache::get_or_populate_in_memory(
+            state.store.get_cache_store().as_ref(),
+            key,
+            func,
+            &cache::ELIMINATION_BASED_DYNAMIC_ALGORITHM_CACHE,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("unable to populate ELIMINATION_BASED_DYNAMIC_ALGORITHM_CACHE")
+    }
+}
+
 /// Cfetch dynamic routing configs
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 #[instrument(skip_all)]
@@ -964,6 +1002,99 @@ pub async fn push_metrics_with_update_window_for_success_based_routing(
     }
 }
 
+/// update window for elimination based dynamic routing
+#[cfg(all(feature = "v1", feature = "dynamic_routing"))]
+#[instrument(skip_all)]
+pub async fn update_window_for_elimination_routing(
+    state: &SessionState,
+    payment_attempt: &storage::PaymentAttempt,
+    profile_id: &id_type::ProfileId,
+    dynamic_algo_ref: routing_types::DynamicRoutingAlgorithmRef,
+    elimination_routing_configs_params_interpolator: DynamicRoutingConfigParamsInterpolator,
+    gsm_error_category: common_enums::ErrorCategory,
+) -> RouterResult<()> {
+    if let Some(elimination_algo_ref) = dynamic_algo_ref.elimination_routing_algorithm {
+        if elimination_algo_ref.enabled_feature != routing_types::DynamicRoutingFeatures::None {
+            let client = state
+                .grpc_client
+                .dynamic_routing
+                .elimination_based_client
+                .as_ref()
+                .ok_or(errors::ApiErrorResponse::GenericNotFoundError {
+                    message: "elimination_rate gRPC client not found".to_string(),
+                })?;
+
+            let elimination_routing_config = fetch_dynamic_routing_configs::<
+                routing_types::EliminationRoutingConfig,
+            >(
+                state,
+                profile_id,
+                elimination_algo_ref
+                    .algorithm_id_with_timestamp
+                    .algorithm_id
+                    .ok_or(errors::ApiErrorResponse::GenericNotFoundError {
+                        message: "elimination routing algorithm_id not found".to_string(),
+                    })
+                    .attach_printable(
+                        "elimination_routing_algorithm_id not found in business_profile",
+                    )?,
+            )
+            .await
+            .change_context(errors::ApiErrorResponse::GenericNotFoundError {
+                message: "elimination based dynamic routing configs not found".to_string(),
+            })
+            .attach_printable("unable to retrieve success_rate based dynamic routing configs")?;
+
+            let payment_connector = &payment_attempt.connector.clone().ok_or(
+                errors::ApiErrorResponse::GenericNotFoundError {
+                    message: "unable to derive payment connector from payment attempt".to_string(),
+                },
+            )?;
+
+            let elimination_routing_config_params = elimination_routing_configs_params_interpolator
+                .get_string_val(
+                    elimination_routing_config
+                        .params
+                        .as_ref()
+                        .ok_or(errors::RoutingError::EliminationBasedRoutingParamsNotFoundError)
+                        .change_context(errors::ApiErrorResponse::InternalServerError)?,
+                );
+
+            client
+                .update_elimination_bucket_config(
+                    profile_id.get_string_repr().to_string(),
+                    elimination_routing_config_params,
+                    vec![routing_types::RoutableConnectorChoiceWithBucketName::new(
+                        routing_types::RoutableConnectorChoice {
+                            choice_kind: api_models::routing::RoutableChoiceKind::FullStruct,
+                            connector: common_enums::RoutableConnectors::from_str(
+                                payment_connector.as_str(),
+                            )
+                            .change_context(errors::ApiErrorResponse::InternalServerError)
+                            .attach_printable(
+                                "unable to infer routable_connector from connector",
+                            )?,
+                            merchant_connector_id: payment_attempt.merchant_connector_id.clone(),
+                        },
+                        gsm_error_category.to_string(),
+                    )],
+                    elimination_routing_config.elimination_analyser_config,
+                    state.get_grpc_headers(),
+                )
+                .await
+                .change_context(errors::ApiErrorResponse::InternalServerError)
+                .attach_printable(
+                    "unable to update elimination based routing buckets in dynamic routing service",
+                )?;
+            Ok(())
+        } else {
+            Ok(())
+        }
+    } else {
+        Ok(())
+    }
+}
+
 /// metrics for contract based dynamic routing
 #[cfg(all(feature = "v1", feature = "dynamic_routing"))]
 #[instrument(skip_all)]