feat(payouts): Implement Smart Retries for Payout

crates/diesel_models/src/payouts.rs

@@ -29,6 +29,7 @@ pub struct Payouts {
     pub created_at: PrimitiveDateTime,
     #[serde(with = "common_utils::custom_serde::iso8601")]
     pub last_modified_at: PrimitiveDateTime,
+    pub attempt_count: i16,
 }
 
 impl Default for Payouts {
@@ -53,6 +54,7 @@ impl Default for Payouts {
             metadata: Option::default(),
             created_at: now,
             last_modified_at: now,
+            attempt_count: i16::default(),
         }
     }
 }
@@ -90,6 +92,7 @@ pub struct PayoutsNew {
     pub created_at: Option<PrimitiveDateTime>,
     #[serde(default, with = "common_utils::custom_serde::iso8601::option")]
     pub last_modified_at: Option<PrimitiveDateTime>,
+    pub attempt_count: i16,
 }
 
 #[derive(Debug)]
@@ -105,6 +108,7 @@ pub enum PayoutsUpdate {
         entity_type: storage_enums::PayoutEntityType,
         metadata: Option<pii::SecretSerdeValue>,
         last_modified_at: Option<PrimitiveDateTime>,
+        attempt_count: i16,
     },
     PayoutMethodIdUpdate {
         payout_method_id: Option<String>,
@@ -130,6 +134,7 @@ pub struct PayoutsUpdateInternal {
     pub metadata: Option<pii::SecretSerdeValue>,
     pub last_modified_at: Option<PrimitiveDateTime>,
     pub payout_method_id: Option<String>,
+    pub attempt_count: Option<i16>,
 }
 
 impl From<PayoutsUpdate> for PayoutsUpdateInternal {
@@ -146,6 +151,7 @@ impl From<PayoutsUpdate> for PayoutsUpdateInternal {
                 entity_type,
                 metadata,
                 last_modified_at,
+                attempt_count,
             } => Self {
                 amount: Some(amount),
                 destination_currency: Some(destination_currency),
@@ -157,6 +163,7 @@ impl From<PayoutsUpdate> for PayoutsUpdateInternal {
                 entity_type: Some(entity_type),
                 metadata,
                 last_modified_at,
+                attempt_count: Some(attempt_count),
                 ..Default::default()
             },
             PayoutsUpdate::PayoutMethodIdUpdate {

crates/diesel_models/src/query/payout_attempt.rs

@@ -34,6 +34,20 @@ impl PayoutAttempt {
         .await
     }
 
+    pub async fn find_by_merchant_id_payout_attempt_id(
+        conn: &PgPooledConn,
+        merchant_id: &str,
+        payout_attempt_id: &str,
+    ) -> StorageResult<Self> {
+        generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
+            conn,
+            dsl::merchant_id
+                .eq(merchant_id.to_owned())
+                .and(dsl::payout_attempt_id.eq(payout_attempt_id.to_owned())),
+        )
+        .await
+    }
+
     pub async fn update_by_merchant_id_payout_id(
         conn: &PgPooledConn,
         merchant_id: &str,
@@ -54,4 +68,25 @@ impl PayoutAttempt {
             report!(errors::DatabaseError::NotFound).attach_printable("Error while updating payout")
         })
     }
+
+    pub async fn update_by_merchant_id_payout_attempt_id(
+        conn: &PgPooledConn,
+        merchant_id: &str,
+        payout_attempt_id: &str,
+        payout: PayoutAttemptUpdate,
+    ) -> StorageResult<Self> {
+        generics::generic_update_with_results::<<Self as HasTable>::Table, _, _, _>(
+            conn,
+            dsl::merchant_id
+                .eq(merchant_id.to_owned())
+                .and(dsl::payout_attempt_id.eq(payout_attempt_id.to_owned())),
+            PayoutAttemptUpdateInternal::from(payout),
+        )
+        .await?
+        .first()
+        .cloned()
+        .ok_or_else(|| {
+            report!(errors::DatabaseError::NotFound).attach_printable("Error while updating payout")
+        })
+    }
 }

crates/diesel_models/src/schema.rs

@@ -897,6 +897,7 @@ diesel::table! {
         metadata -> Nullable<Jsonb>,
         created_at -> Timestamp,
         last_modified_at -> Timestamp,
+        attempt_count -> Int2,
     }
 }
 

crates/router/Cargo.toml

@@ -9,7 +9,7 @@ readme = "README.md"
 license.workspace = true
 
 [features]
-default = ["kv_store", "stripe", "oltp", "olap", "backwards_compatibility", "accounts_cache", "dummy_connector", "payouts", "business_profile_routing", "connector_choice_mca_id", "profile_specific_fallback_routing", "retry", "frm"]
+default = ["kv_store", "stripe", "oltp", "olap", "backwards_compatibility", "accounts_cache", "dummy_connector", "payouts", "payout_retry", "business_profile_routing", "connector_choice_mca_id", "profile_specific_fallback_routing", "retry", "frm"]
 aws_s3 = ["external_services/aws_s3"]
 kms = ["external_services/kms"]
 hashicorp-vault = ["external_services/hashicorp-vault"]
@@ -31,6 +31,7 @@ connector_choice_mca_id = ["api_models/connector_choice_mca_id", "euclid/connect
 external_access_dc = ["dummy_connector"]
 detailed_errors = ["api_models/detailed_errors", "error-stack/serde"]
 payouts = []
+payout_retry = []
 recon = ["email", "api_models/recon"]
 retry = []
 

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

@@ -773,7 +773,6 @@ pub async fn perform_eligibility_analysis_with_fallback<F: Clone>(
         profile_id.clone(),
     )
     .await?;
-
     let fallback_selection = perform_fallback_routing(
         state,
         key_store,
@@ -784,7 +783,6 @@ pub async fn perform_eligibility_analysis_with_fallback<F: Clone>(
         profile_id,
     )
     .await;
-
     final_selection.append(
         &mut fallback_selection
             .unwrap_or_default()