feat(payouts): Implement Single Connector Retry for Payouts

crates/diesel_models/src/gsm.rs

@@ -14,6 +14,8 @@ use crate::schema::gateway_status_map;
     Debug,
     Eq,
     PartialEq,
+    Ord,
+    PartialOrd,
     router_derive::DebugAsDisplay,
     Identifiable,
     Queryable,

crates/router/src/core/payouts.rs

@@ -150,15 +150,40 @@ pub async fn make_connector_decision(
 ) -> RouterResult<PayoutData> {
     match connector_call_type {
         api::ConnectorCallType::PreDetermined(connector_data) => {
-            call_connector_payout(
+            payout_data = call_connector_payout(
                 state,
                 merchant_account,
                 key_store,
                 req,
                 &connector_data,
                 &mut payout_data,
             )
-            .await
+            .await?;
+
+            #[cfg(feature = "payout_retry")]
+            {
+                use crate::core::payouts::retry::{self, GsmValidation};
+                let config_bool = retry::config_should_call_gsm_payout(
+                    &*state.store,
+                    &merchant_account.merchant_id,
+                    retry::PayoutRetryType::SingleConnector,
+                )
+                .await;
+
+                if config_bool && payout_data.should_call_gsm() {
+                    payout_data = retry::do_gsm_single_connector_actions(
+                        state,
+                        connector_data,
+                        payout_data,
+                        merchant_account,
+                        key_store,
+                        req,
+                    )
+                    .await?;
+                }
+            }
+
+            Ok(payout_data)
         }
         api::ConnectorCallType::Retryable(connectors) => {
             let mut connectors = connectors.into_iter();
@@ -178,16 +203,36 @@ pub async fn make_connector_decision(
             #[cfg(feature = "payout_retry")]
             {
                 use crate::core::payouts::retry::{self, GsmValidation};
-                let config_bool = retry::config_should_call_gsm_payout(
+                let config_multiple_connector_bool = retry::config_should_call_gsm_payout(
                     &*state.store,
                     &merchant_account.merchant_id,
+                    retry::PayoutRetryType::MultiConnector,
                 )
                 .await;
 
-                if config_bool && payout_data.should_call_gsm() {
-                    payout_data = retry::do_gsm_actions(
+                if config_multiple_connector_bool && payout_data.should_call_gsm() {
+                    payout_data = retry::do_gsm_multiple_connector_actions(
                         state,
                         connectors,
+                        connector_data.clone(),
+                        payout_data,
+                        merchant_account,
+                        key_store,
+                        req,
+                    )
+                    .await?;
+                }
+
+                let config_single_connector_bool = retry::config_should_call_gsm_payout(
+                    &*state.store,
+                    &merchant_account.merchant_id,
+                    retry::PayoutRetryType::SingleConnector,
+                )
+                .await;
+
+                if config_single_connector_bool && payout_data.should_call_gsm() {
+                    payout_data = retry::do_gsm_single_connector_actions(
+                        state,
                         connector_data,
                         payout_data,
                         merchant_account,
@@ -349,74 +394,52 @@ pub async fn payouts_update_core(
         }
     }
 
-    payout_data = match (
-        req.connector.clone(),
-        payout_data.payout_attempt.connector.clone(),
-    ) {
+    if (
+        req.connector.is_none(),
+        payout_data.payout_attempt.connector.is_some(),
+    ) != (true, true)
+    {
         // if the connector is not updated but was provided during payout create
-        (None, Some(connector)) => {
-            let connector_data = api::ConnectorData::get_payout_connector_by_name(
-                &state.conf.connectors,
-                connector.as_str(),
-                api::GetToken::Connector,
-                payout_attempt.merchant_connector_id.clone(),
-            )
-            .change_context(errors::ApiErrorResponse::InternalServerError)
-            .attach_printable("Failed to get the connector data")?;
+        payout_data.payout_attempt.connector = None;
+        payout_data.payout_attempt.routing_info = None;
 
-            call_connector_payout(
+        //fetch payout_method_data
+        payout_data.payout_method_data = Some(
+            helpers::make_payout_method_data(
                 &state,
-                &merchant_account,
+                req.payout_method_data.as_ref(),
+                payout_data.payout_attempt.payout_token.as_deref(),
+                &payout_data.payout_attempt.customer_id,
+                &payout_data.payout_attempt.merchant_id,
+                &payout_data.payout_attempt.payout_id,
+                Some(&payouts.payout_type),
                 &key_store,
-                &req,
-                &connector_data,
-                &mut payout_data,
             )
             .await?
-        }
-        // if the connector is updated or not present both in create and update call
-        _ => {
-            payout_data.payout_attempt.connector = None;
-            payout_data.payout_attempt.routing_info = None;
-
-            //fetch payout_method_data
-            payout_data.payout_method_data = Some(
-                helpers::make_payout_method_data(
-                    &state,
-                    req.payout_method_data.as_ref(),
-                    payout_data.payout_attempt.payout_token.as_deref(),
-                    &payout_data.payout_attempt.customer_id,
-                    &payout_data.payout_attempt.merchant_id,
-                    &payout_data.payout_attempt.payout_id,
-                    Some(&payouts.payout_type),
-                    &key_store,
-                )
-                .await?
-                .get_required_value("payout_method_data")?,
-            );
+            .get_required_value("payout_method_data")?,
+        );
+    };
 
-            let connector_call_type = get_connector_choice(
-                &state,
-                &merchant_account,
-                &key_store,
-                None,
-                req.routing.clone(),
-                &mut payout_data,
-                req.connector.clone(),
-            )
-            .await?;
+    let connector_call_type = get_connector_choice(
+        &state,
+        &merchant_account,
+        &key_store,
+        None,
+        req.routing.clone(),
+        &mut payout_data,
+        req.connector.clone(),
+    )
+    .await?;
 
-            make_connector_decision(
-                &state,
-                &merchant_account,
-                &key_store,
-                &req,
-                connector_call_type,
-                payout_data,
-            )
-            .await?
-        }
-    };
+    payout_data = make_connector_decision(
+        &state,
+        &merchant_account,
+        &key_store,
+        &req,
+        connector_call_type,
+        payout_data,
+    )
+    .await?;
 
     response_handler(
         &state,

crates/router/src/core/payouts/retry.rs

@@ -1,4 +1,4 @@
-use std::{str::FromStr, vec::IntoIter};
+use std::{cmp::Ordering, str::FromStr, vec::IntoIter};
 
 use api_models::payouts::PayoutCreateRequest;
 use error_stack::{IntoReport, ResultExt};
@@ -19,9 +19,16 @@ use crate::{
     utils,
 };
 
+#[derive(Clone, Debug, serde::Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum PayoutRetryType {
+    SingleConnector,
+    MultiConnector,
+}
+
 #[instrument(skip_all)]
 #[allow(clippy::too_many_arguments)]
-pub async fn do_gsm_actions(
+pub async fn do_gsm_multiple_connector_actions(
     state: &app::AppState,
     mut connectors: IntoIter<api::ConnectorData>,
     original_connector_data: api::ConnectorData,
@@ -41,7 +48,13 @@ pub async fn do_gsm_actions(
 
         match get_gsm_decision(gsm) {
             api_models::gsm::GsmDecision::Retry => {
-                retries = get_retries(state, retries, &merchant_account.merchant_id).await;
+                retries = get_retries(
+                    state,
+                    retries,
+                    &merchant_account.merchant_id,
+                    PayoutRetryType::MultiConnector,
+                )
+                .await;
 
                 if retries.is_none() || retries == Some(0) {
                     metrics::AUTO_PAYOUT_RETRY_EXHAUSTED_COUNT.add(&metrics::CONTEXT, 1, &[]);
@@ -83,16 +96,91 @@ pub async fn do_gsm_actions(
     Ok(payout_data)
 }
 
+#[instrument(skip_all)]
+#[allow(clippy::too_many_arguments)]
+pub async fn do_gsm_single_connector_actions(
+    state: &app::AppState,
+    original_connector_data: api::ConnectorData,
+    mut payout_data: PayoutData,
+    merchant_account: &domain::MerchantAccount,
+    key_store: &domain::MerchantKeyStore,
+    req: &PayoutCreateRequest,
+) -> RouterResult<PayoutData> {
+    let mut retries = None;
+
+    metrics::AUTO_PAYOUT_RETRY_ELIGIBLE_REQUEST_COUNT.add(&metrics::CONTEXT, 1, &[]);
+
+    let mut previous_gsm = None; // to compare previous status
+
+    loop {
+        let gsm = get_gsm(state, &original_connector_data, &payout_data).await?;
+
+        // if the error config is same as previous, we break out of the loop
+        if let Ordering::Equal = gsm.cmp(&previous_gsm) {
+            break;
+        }
+        previous_gsm = gsm.clone();
+
+        match get_gsm_decision(gsm) {
+            api_models::gsm::GsmDecision::Retry => {
+                retries = get_retries(
+                    state,
+                    retries,
+                    &merchant_account.merchant_id,
+                    PayoutRetryType::SingleConnector,
+                )
+                .await;
+
+                if retries.is_none() || retries == Some(0) {
+                    metrics::AUTO_PAYOUT_RETRY_EXHAUSTED_COUNT.add(&metrics::CONTEXT, 1, &[]);
+                    logger::info!("retries exhausted for auto_retry payment");
+                    break;
+                }
+
+                payout_data = do_retry(
+                    &state.clone(),
+                    original_connector_data.to_owned(),
+                    merchant_account,
+                    key_store,
+                    payout_data,
+                    req,
+                )
+                .await?;
+
+                retries = retries.map(|i| i - 1);
+            }
+            api_models::gsm::GsmDecision::Requeue => {
+                Err(errors::ApiErrorResponse::NotImplemented {
+                    message: errors::api_error_response::NotImplementedMessage::Reason(
+                        "Requeue not implemented".to_string(),
+                    ),
+                })
+                .into_report()?
+            }
+            api_models::gsm::GsmDecision::DoDefault => break,
+        }
+    }
+    Ok(payout_data)
+}
+
 #[instrument(skip_all)]
 pub async fn get_retries(
     state: &app::AppState,
     retries: Option<i32>,
     merchant_id: &str,
+    retry_type: PayoutRetryType,
 ) -> Option<i32> {
     match retries {
         Some(retries) => Some(retries),
         None => {
-            let key = format!("max_auto_payout_retries_enabled_{merchant_id}");
+            let key = match retry_type {
+                PayoutRetryType::SingleConnector => {
+                    format!("max_auto_single_connector_payout_retries_enabled_{merchant_id}")
+                }
+                PayoutRetryType::MultiConnector => {
+                    format!("max_auto_multiple_connector_payout_retries_enabled_{merchant_id}")
+                }
+            };
             let db = &*state.store;
             db.find_config_by_key(key.as_str())
                 .await
@@ -235,12 +323,18 @@ pub async fn modify_trackers(
 pub async fn config_should_call_gsm_payout(
     db: &dyn StorageInterface,
     merchant_id: &String,
+    retry_type: PayoutRetryType,
 ) -> bool {
+    let key = match retry_type {
+        PayoutRetryType::SingleConnector => {
+            format!("should_call_gsm_single_connector_payout_{}", merchant_id)
+        }
+        PayoutRetryType::MultiConnector => {
+            format!("should_call_gsm_multiple_connector_payout_{}", merchant_id)
+        }
+    };
     let config = db
-        .find_config_by_key_unwrap_or(
-            format!("should_call_gsm_payout_{}", merchant_id).as_str(),
-            Some("false".to_string()),
-        )
+        .find_config_by_key_unwrap_or(key.as_str(), Some("false".to_string()))
         .await;
     match config {
         Ok(conf) => conf.config == "true",