feat: Fix global elimination redis key issue (#198)

Author: PriyanshuC132

Cargo.lock

@@ -9,7 +9,7 @@ rust-version = "1.85.0"
 
 [features]
 default = ["middleware","mysql"]
-release = ["middleware", "kms-aws","mysql"]
+release = ["middleware", "kms-aws","mysql", "redis_compression"]
 kms-aws = ["dep:aws-config", "dep:aws-sdk-kms"]
 kms-hashicorp-vault = ["dep:vaultrs"]
 limit = []
@@ -19,6 +19,7 @@ external_key_manager = []
 external_key_manager_mtls = ["external_key_manager", "reqwest/rustls-tls"]
 postgres = []
 mysql = []
+redis_compression = ["dep:zstd"]
 
 [dependencies]
 aws-config = { version = "1.5.5", optional = true }
@@ -36,6 +37,7 @@ jemallocator = "0.5"
 jemalloc-ctl = "0.5"
 prometheus = "0.13"
 lazy_static = "1.4"
+zstd = { version = "0.12", optional = true }
 
 async-bb8-diesel = { git = "https://github.com/jarnura/async-bb8-diesel", rev = "53b4ab901aab7635c8215fd1c2d542c8db443094" }
 redis_interface = { git = "https://github.com/juspay/hyperswitch.git", tag = "2024.09.30.0" }

Cargo.toml

@@ -53,6 +53,9 @@ max_capacity = 5000
 service_config_redis_prefix = "DE_service_config_"
 service_config_ttl = 300    # 5 minutes
 
+[compression_filepath]
+zstd_compression_filepath = "/tmp/extra-paths/redis-zstd-dictionaries/sbx"
+
 [tenant_secrets]
 public = { schema = "public" }
 

config/development.toml

@@ -71,7 +71,10 @@ impl TenantAppState {
 
         Ok(Self {
             db,
-            redis_conn: Arc::new(RedisConnectionWrapper::new(redis_conn)),
+            redis_conn: Arc::new(RedisConnectionWrapper::new(
+                redis_conn,
+                global_config.compression_filepath.clone(),
+            )),
             api_client,
             config: tenant_config,
         })

src/app.rs

@@ -42,6 +42,7 @@ pub struct GlobalConfig {
     pub routing_config: Option<TomlConfig>,
     #[serde(default)]
     pub debit_routing_config: network_decider::types::DebitRoutingConfig,
+    pub compression_filepath: Option<CompressionFilepath>,
 }
 
 #[derive(Clone, Debug)]
@@ -121,6 +122,11 @@ impl CacheConfig {
     }
 }
 
+#[derive(Clone, serde::Deserialize, Debug)]
+pub struct CompressionFilepath {
+    pub zstd_compression_filepath: String,
+}
+
 #[derive(serde::Deserialize, Debug, Clone)]
 pub struct TenantSecrets {
     /// schema name for the tenant (defaults to tenant_id)

src/config.rs

@@ -22,6 +22,7 @@ use super::utils as Utils;
 // use optics_core::{preview, review};
 use crate::decider::gatewaydecider::constants as C;
 use crate::logger;
+use crate::redis::feature::RedisDataStruct;
 use crate::types::card::txn_card_info::TxnCardInfo;
 use crate::types::merchant as ETM;
 use crate::types::merchant::merchant_gateway_account::MerchantGatewayAccount;
@@ -71,7 +72,7 @@ pub async fn decider_full_payload_hs_function(
         Some(card_bin) => Some(card_bin),
         None => match dreq.txnCardInfo.card_isin {
             Some(c_isin) => {
-                let res_bin = Utils::get_card_bin_from_token_bin(6, c_isin.as_str()).await;
+                let res_bin = Utils::get_card_bin_from_token_bin(6, c_isin.as_str(), None).await;
                 Some(res_bin)
             }
             None => dreq.txnCardInfo.card_isin.clone(),
@@ -105,6 +106,7 @@ pub async fn decider_full_payload_hs_function(
         dpPriorityLogicScript: dreq.priorityLogicScript,
         dpEDCCApplied: dreq.isEdccApplied,
         dpShouldConsumeResult: dreq.shouldConsumeResult,
+        dpRedisCompressionConfig: None,
     };
 
     if dreq_.ranking_algorithm == Some(RankingAlgorithm::NtwBasedRouting) {
@@ -534,6 +536,8 @@ pub async fn run_decider_flow(
                 .unwrap_or_default()
                 .as_str(),
             C::GATEWAY_SCORE_KEYS_TTL,
+            None,
+            RedisDataStruct::STRING,
         )
         .await
         .unwrap_or_default();

src/decider/gatewaydecider/flow_new.rs

@@ -1,5 +1,6 @@
 use super::runner::get_gateway_priority;
 use super::types::UnifiedError;
+use crate::redis::feature::{RedisCompressionConfigCombined, RedisDataStruct};
 use axum::response::IntoResponse;
 use serde_json::json;
 use serde_json::Value as AValue;
@@ -127,6 +128,7 @@ pub trait ResponseDecider {
 
 pub async fn decider_full_payload_hs_function(
     dreq: T::DomainDeciderRequest,
+    redis_compression_config: Option<RedisCompressionConfigCombined>,
 ) -> Result<(T::DecidedGateway, Vec<(String, Vec<String>)>), T::ErrorResponse> {
     let merchant_prefs = match ETM::merchant_iframe_preferences::getMerchantIPrefsByMId(
         dreq.txnDetail.merchantId.0.clone(),
@@ -165,7 +167,12 @@ pub async fn decider_full_payload_hs_function(
         Some(card_bin) => Some(card_bin),
         None => match dreq.txnCardInfo.card_isin {
             Some(c_isin) => {
-                let res_bin = Utils::get_card_bin_from_token_bin(6, c_isin.as_str()).await;
+                let res_bin = Utils::get_card_bin_from_token_bin(
+                    6,
+                    c_isin.as_str(),
+                    redis_compression_config.clone(),
+                )
+                .await;
                 Some(res_bin)
             }
             None => dreq.txnCardInfo.card_isin.clone(),
@@ -199,6 +206,7 @@ pub async fn decider_full_payload_hs_function(
         dpPriorityLogicScript: dreq.priorityLogicScript,
         dpEDCCApplied: dreq.isEdccApplied,
         dpShouldConsumeResult: dreq.shouldConsumeResult,
+        dpRedisCompressionConfig: redis_compression_config,
     };
     run_decider_flow(decider_params, true).await
 }
@@ -698,6 +706,8 @@ pub async fn run_decider_flow(
                     .unwrap_or_default()
                     .as_str(),
                 C::GATEWAY_SCORE_KEYS_TTL,
+                deciderParams.dpRedisCompressionConfig.clone(),
+                RedisDataStruct::STRING,
             )
             .await
             .unwrap_or_default();

src/decider/gatewaydecider/flows.rs

@@ -30,7 +30,7 @@ use rand_distr::{Beta, Binomial, Distribution};
 use serde::{Deserialize, Serialize};
 use time::{OffsetDateTime, PrimitiveDateTime};
 // use crate::types::card_brand_routes as ETCBR;
-use crate::redis::feature::{self as M, is_feature_enabled};
+use crate::redis::feature::{self as M, is_feature_enabled, RedisCompressionConfigCombined};
 use crate::types::gateway_routing_input as ETGRI;
 // use crate::types::gateway_health as ETGH;
 use crate::types::card as ETCT;
@@ -3009,6 +3009,7 @@ pub async fn trigger_reset_gateway_score(
                     decider_flow,
                     txn_detail.clone(),
                     reset_gateway_input.clone(),
+                    decider_flow.get().dpRedisCompressionConfig.clone(),
                 )
                 .await;
                 reset_gateway_sr_list.push(reset_gateway_input.clone());
@@ -3071,6 +3072,7 @@ pub async fn reset_gateway_score(
     decider_flow: &mut DeciderFlow<'_>,
     txn_detail: ETTD::TxnDetail,
     reset_gateway_input: ResetGatewayInput,
+    redis_compression_config: Option<RedisCompressionConfigCombined>,
 ) {
     let current_timestamp = get_current_date_in_millis();
     match (
@@ -3081,6 +3083,14 @@ pub async fn reset_gateway_score(
         (Some(key), Some(threshold), Some(max_count)) => {
             let penality_factor = Utils::get_penality_factor_(decider_flow).await;
             let score = get_merchant_elimination_gateway_score(key.clone()).await;
+            logger::debug!(
+                tag = "scoringFlow",
+                action = "scoringFlow",
+                "Current Gateway Score for {:?} : key {:?} before reset attempt: {:?}",
+                txn_detail.txnId,
+                key,
+                score
+            );
             let (is_eligible_for_reset, reset_cached_gateway_score) = match score {
                 Some(score) => {
                     let current_score = score.score;
@@ -3141,6 +3151,7 @@ pub async fn reset_gateway_score(
                 key.clone(),
                 reset_cached_gateway_score.clone(),
                 safe_remaining_ttl,
+                redis_compression_config,
             )
             .await;
             match result {

src/decider/gatewaydecider/gw_scoring.rs

@@ -509,7 +509,9 @@ pub async fn execute_priority_logic(
     let resolve_bin = match utils::fetch_extended_card_bin(&req.txnCardInfo) {
         Some(card_bin) => Some(card_bin),
         None => match req.txnCardInfo.card_isin {
-            Some(c_isin) => Some(utils::get_card_bin_from_token_bin(6, c_isin.as_str()).await),
+            Some(c_isin) => {
+                Some(utils::get_card_bin_from_token_bin(6, c_isin.as_str(), None).await)
+            }
             None => None,
         },
     };

src/decider/gatewaydecider/runner.rs

@@ -1,5 +1,6 @@
 use crate::app::{get_tenant_app_state, TenantAppState};
 use crate::decider::network_decider;
+use crate::redis::feature::RedisCompressionConfigCombined;
 use crate::types::country::country_iso::CountryISO2;
 use crate::types::currency::Currency;
 use crate::types::money::internal as ETMo;
@@ -1297,6 +1298,7 @@ pub struct DeciderParams {
     pub dpPriorityLogicScript: Option<String>,
     pub dpEDCCApplied: Option<bool>,
     pub dpShouldConsumeResult: Option<bool>,
+    pub dpRedisCompressionConfig: Option<RedisCompressionConfigCombined>,
 }
 
 #[derive(Debug, Serialize, Deserialize)]