add Redis support and configuration

su-shivanshmathur · su-shivanshmathur · commit 8e3eb792d72a · 2026-06-17T09:23:38.000+05:30
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -9,24 +9,25 @@ rust-version = "1.85"
 
 [features]
 default = ["caching"]
-release = ["kms-aws", "middleware", "key_custodian", "limit", "kms-hashicorp-vault", "caching", "external_key_manager", "vergen"]
+release = ["kms-aws", "middleware", "key_custodian", "limit", "kms-hashicorp-vault", "caching", "external_key_manager", "vergen", "redis"]
 kms-aws = ["dep:aws-config", "dep:aws-sdk-kms"]
 kms-hashicorp-vault = ["dep:vaultrs"]
 limit = []
 middleware = []
 key_custodian = []
 caching = ["dep:moka"]
+redis = ["dep:hyperswitch_redis_interface"]
 console = ["tokio/tracing", "dep:console-subscriber"]
 external_key_manager = ["reqwest/rustls-tls"]
 vergen = ["build_info/vergen-gix"]
 
 [dependencies]
-async-trait = "0.1.81"
+async-trait = "0.1.88"
 aws-config = { version = "1.5.5", optional = true }
 aws-sdk-kms = { version = "1.40.0", optional = true }
 base64 = "0.22.1"
 bytes = "1.7.1"
-futures = "0.3.30"
+futures = "0.3.31"
 gethostname = "0.5.0"
 rustc-hash = "2.1"
 once_cell = "1.19.0"
@@ -46,24 +47,25 @@ tracing-subscriber = { version = "0.3.18", default-features = true, features = [
 console-subscriber = { version = "0.4.0", optional = true }
 http-body-util = "0.1.2"
 
-diesel = { version = "2.2.3", features = ["postgres", "serde_json", "time"] }
+diesel = { version = "2.2.10", features = ["postgres", "serde_json", "time"] }
 diesel-async = { version = "0.5.0", features = ["postgres", "deadpool"] }
 
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.140"
 josekit = "0.8.7"
 
-thiserror = "1.0.63"
+thiserror = "1.0.69"
 config = "0.14.0"
 serde_path_to_error = "0.1.16"
 error-stack = "0.5.0"
 futures-util = "0.3.30"
 digest = "0.10"
 hyperswitch_masking = { version = "0.0.1", features = ["diesel", "serde"] }
-ring = { version = "0.17.8", features = ["std"] }
+hyperswitch_redis_interface = { git = "https://github.com/juspay/hyperswitch", tag = "2026.06.12.0", package = "redis_interface", default-features = false, features = ["fred"], optional = true }
+ring = { version = "0.17.14", features = ["std"] }
 hex = "0.4.3"
 time = { version = "0.3.45", features = ["serde"] }
-uuid = { version = "1.10.0", features = ["v7", "fast-rng"] }
+uuid = { version = "1.20.0", features = ["v7", "fast-rng"] }
 moka = { version = "0.12.8", features = ["future"], optional = true }
 reqwest = { version = "0.12.7", features = ["json", "__rustls"] }
 
diff --git a/config/config.example.toml b/config/config.example.toml
@@ -83,3 +83,24 @@ mode = "disabled"  # Options: "disabled", "enabled", "enabled_with_mtls"
 # -----END CERTIFICATE-----
 # """
 
+# Redis configuration (only used with the `redis` feature)
+[redis]
+host = "127.0.0.1"                 # Redis server host
+port = 6379                        # Redis server port
+pool_size = 5                      # number of connections in the pool
+reconnect_max_attempts = 0         # max reconnection attempts (0 = retry forever)
+reconnect_delay = 200              # delay between reconnection attempts (milliseconds)
+default_ttl = 300                  # default key TTL (seconds)
+default_hash_ttl = 900             # default hash-table TTL (seconds)
+use_legacy_version = false         # use RESP2 instead of RESP3
+stream_read_count = 1              # entries read per stream read
+auto_pipeline = true               # automatically pipeline commands
+broadcast_channel_capacity = 32    # pub/sub broadcast channel capacity
+disable_auto_backpressure = false  # disable automatic backpressure
+max_in_flight_commands = 5000      # max in-flight commands before backpressure (0 = disabled)
+default_command_timeout = 30       # command timeout (seconds)
+unresponsive_timeout = 10          # mark a connection unresponsive after (seconds)
+unresponsive_check_interval = 2    # how often to check for unresponsiveness (seconds)
+max_feed_count = 200               # max commands fed to the server at once
+max_failure_threshold_seconds = 5  # max seconds Redis may be unreachable before it's failed
+
diff --git a/config/development.toml b/config/development.toml
@@ -30,3 +30,23 @@ public = { master_key = "feffe9928665731c6d6a8f9467308308feffe9928665731c6d6a8f9
 
 [external_key_manager]
 mode = "disabled"
+
+[redis]
+host = "127.0.0.1"
+port = 6379
+pool_size = 5
+reconnect_max_attempts = 0
+reconnect_delay = 200
+default_ttl = 300
+default_hash_ttl = 900
+use_legacy_version = false
+stream_read_count = 1
+auto_pipeline = true
+broadcast_channel_capacity = 32
+disable_auto_backpressure = false
+max_in_flight_commands = 5000
+default_command_timeout = 30
+unresponsive_timeout = 10
+unresponsive_check_interval = 2
+max_feed_count = 200
+max_failure_threshold_seconds = 5
diff --git a/config/docker-configuration.toml b/config/docker-configuration.toml
@@ -32,3 +32,23 @@ identity = ""
 [external_key_manager]
 url = "http://localhost:5000"
 cert = ""
+
+[redis]
+host = "127.0.0.1"
+port = 6379
+pool_size = 5
+reconnect_max_attempts = 0
+reconnect_delay = 200
+default_ttl = 300
+default_hash_ttl = 900
+use_legacy_version = false
+stream_read_count = 1
+auto_pipeline = true
+broadcast_channel_capacity = 32
+disable_auto_backpressure = false
+max_in_flight_commands = 5000
+default_command_timeout = 30
+unresponsive_timeout = 10
+unresponsive_check_interval = 2
+max_feed_count = 200
+max_failure_threshold_seconds = 5
diff --git a/src/app.rs b/src/app.rs
@@ -43,6 +43,8 @@ pub struct TenantAppState {
     pub db: Storage,
     pub config: config::TenantConfig,
     pub api_client: ApiClient,
+    #[cfg(feature = "redis")]
+    pub redis: Option<storage::redis::RedisStore>,
 }
 
 #[allow(clippy::expect_used)]
@@ -54,6 +56,8 @@ impl TenantAppState {
         global_config: &GlobalConfig,
         tenant_config: TenantConfig,
         api_client: ApiClient,
+        #[cfg(feature = "redis")] 
+        shared_redis: Option<&storage::redis::RedisStore>,
     ) -> error_stack::Result<Self, error::ConfigurationError> {
         #[allow(clippy::map_identity)]
         let db = storage::Storage::new(
@@ -72,6 +76,8 @@ impl TenantAppState {
         Ok(Self {
             db,
             api_client,
+            #[cfg(feature = "redis")]
+            redis: shared_redis.map(|store| store.clone_with_prefix(&tenant_config.tenant_id)),
             config: tenant_config,
         })
     }
diff --git a/src/config.rs b/src/config.rs
@@ -8,6 +8,8 @@ use error_stack::ResultExt;
 use hyperswitch_masking::ExposeInterface;
 #[cfg(feature = "external_key_manager")]
 use hyperswitch_masking::Secret;
+#[cfg(feature = "redis")]
+use hyperswitch_redis_interface::RedisSettings;
 
 use crate::{
     api_client::ApiClientConfig,
@@ -36,6 +38,8 @@ pub struct GlobalConfig {
     pub api_client: ApiClientConfig,
     #[serde(default)]
     pub external_key_manager: ExternalKeyManagerConfig,
+    #[cfg(feature = "redis")]
+    pub redis: Option<RedisSettings>,
 }
 
 #[derive(Clone, Debug)]
diff --git a/src/routes/health.rs b/src/routes/health.rs
@@ -36,6 +36,8 @@ pub struct Diagnostics {
     database: DatabaseHealth,
     #[cfg(feature = "external_key_manager")]
     keymanager_status: HealthState,
+    #[cfg(feature = "redis")]
+    redis_status: HealthState,
 }
 
 #[derive(Debug, serde::Serialize, Default)]
@@ -51,7 +53,7 @@ pub enum HealthState {
     Working,
     #[default]
     Failing,
-    #[cfg(feature = "external_key_manager")]
+    #[cfg(any(feature = "external_key_manager", feature = "redis"))]
     Disabled,
 }
 
@@ -109,10 +111,24 @@ pub async fn diagnostics(TenantStateResolver(state): TenantStateResolver) -> Jso
         }
     };
 
+    #[cfg(feature = "redis")]
+    let redis_status = match &state.redis {
+        None => HealthState::Disabled,
+        Some(redis) => match redis.test().await {
+            Ok(()) => HealthState::Working,
+            Err(err) => {
+                crate::logger::error!(redis_err=?err);
+                HealthState::Failing
+            }
+        },
+    };
+
     axum::Json(Diagnostics {
         key_custodian_locked: false,
         database: db_health,
         #[cfg(feature = "external_key_manager")]
         keymanager_status,
+        #[cfg(feature = "redis")]
+        redis_status,
     })
 }
diff --git a/src/routes/key_custodian.rs b/src/routes/key_custodian.rs
@@ -118,6 +118,8 @@ pub async fn decrypt(
                 &global_app_state.global_config,
                 tenant_config,
                 global_app_state.api_client.clone(),
+                #[cfg(feature = "redis")]
+                global_app_state.redis_store.as_ref(),
             )
             .await
             .change_context(error::ApiError::TenantError(
diff --git a/src/storage.rs b/src/storage.rs
@@ -20,6 +20,8 @@ use crate::{
 pub mod caching;
 pub mod consts;
 pub mod db;
+#[cfg(feature = "redis")]
+pub mod redis;
 pub mod schema;
 pub mod storage_v2;
 pub mod types;
diff --git a/src/storage/consts.rs b/src/storage/consts.rs
@@ -15,6 +15,16 @@ pub const X_TENANT_ID: &str = "x-tenant-id";
 pub const X_REQUEST_ID: &str = "x-request-id";
 /// Header key for caller-supplied fingerprint ID (optional)
 pub const X_FINGERPRINT_ID: &str = "x-fingerprint-id";
+/// Key written by the Redis health-check probe
+#[cfg(feature = "redis")]
+pub const REDIS_HEALTH_CHECK_KEY: &str = "health_check_redis";
+/// Value written by the Redis health-check probe
+#[cfg(feature = "redis")]
+pub const REDIS_HEALTH_CHECK_VALUE: &str = "1";
+/// TTL (seconds) on the probe key so Redis drops it even if the delete is skipped
+#[cfg(feature = "redis")]
+pub const REDIS_HEALTH_CHECK_EXPIRY: i64 = 5;
+
 /// Header Constants
 pub mod headers {
     pub const CONTENT_TYPE: &str = "Content-Type";
diff --git a/src/storage/redis.rs b/src/storage/redis.rs
@@ -0,0 +1,80 @@
+use std::sync::Arc;
+
+use hyperswitch_redis_interface::{RedisConnectionPool, RedisSettings, errors::RedisError};
+use tracing::Instrument;
+
+use crate::storage::consts;
+
+// error_stack 0.4 (redis_interface) vs 0.5 (tartarus): rebuild, `?` can't bridge them.
+fn into_report(err: impl std::fmt::Debug) -> error_stack::Report<RedisError> {
+    error_stack::Report::new(RedisError::RedisConnectionError).attach_printable(format!("{err:?}"))
+}
+
+/// A shared `redis_interface` connection pool handle.
+#[derive(Clone)]
+pub struct RedisStore {
+    redis_conn: Arc<RedisConnectionPool>,
+}
+
+impl std::fmt::Debug for RedisStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RedisStore")
+            .field("redis_conn", &"RedisConnectionPool doesn't implement Debug")
+            .finish()
+    }
+}
+
+impl RedisStore {
+    pub async fn new(conf: &RedisSettings) -> error_stack::Result<Self, RedisError> {
+        let pool = RedisConnectionPool::new(conf).await.map_err(into_report)?;
+        Ok(Self {
+            redis_conn: Arc::new(pool),
+        })
+    }
+
+    /// A handle onto the same pool that namespaces every key with `key_prefix`.
+    pub fn clone_with_prefix(&self, key_prefix: &str) -> Self {
+        // `.as_ref().clone(..)` calls the pool's inherent `clone`, not `Arc::clone`.
+        Self {
+            redis_conn: Arc::new(self.redis_conn.as_ref().clone(key_prefix)),
+        }
+    }
+
+    // Logs disconnects via `on_error`. `rx` stays bound (not `_`) so its `tx.send` succeeds.
+    pub fn spawn_error_watcher(&self) {
+        let redis_conn = self.redis_conn.clone();
+        tokio::spawn(
+            async move {
+                let (tx, _rx) = tokio::sync::oneshot::channel();
+                redis_conn.on_error(tx).await;
+            }
+            .in_current_span(),
+        );
+    }
+
+    /// The shared pool. It manages (re)connection internally, so callers run
+    /// commands directly and surface per-command errors themselves.
+    pub fn get_redis_conn(&self) -> Arc<RedisConnectionPool> {
+        self.redis_conn.clone()
+    }
+
+    pub async fn test(&self) -> error_stack::Result<(), RedisError> {
+        let redis_conn = self.get_redis_conn();
+        let key = consts::REDIS_HEALTH_CHECK_KEY.into();
+        redis_conn
+            .set_key_with_expiry(
+                &key,
+                consts::REDIS_HEALTH_CHECK_VALUE,
+                consts::REDIS_HEALTH_CHECK_EXPIRY,
+            )
+            .await
+            .map_err(into_report)?;
+        let value: String = redis_conn.get_key(&key).await.map_err(into_report)?;
+        if value != consts::REDIS_HEALTH_CHECK_VALUE {
+            return Err(error_stack::Report::new(RedisError::UnknownResult)
+                .attach_printable("Redis health-check value mismatch"));
+        }
+        redis_conn.delete_key(&key).await.map_err(into_report)?;
+        Ok(())
+    }
+}
diff --git a/src/tenant.rs b/src/tenant.rs
@@ -16,6 +16,8 @@ pub struct GlobalAppState {
     pub api_client: ApiClient,
     pub known_tenants: HashSet<String>,
     pub global_config: GlobalConfig,
+    #[cfg(feature = "redis")]
+    pub redis_store: Option<crate::storage::redis::RedisStore>,
 }
 
 impl GlobalAppState {
@@ -43,6 +45,25 @@ impl GlobalAppState {
         #[allow(clippy::expect_used)]
         let api_client = ApiClient::new(&global_config).expect("Failed to create api client");
 
+        // Shared pool; tenants derive key-prefixed handles. None if unconfigured or unreachable.
+        #[cfg(feature = "redis")]
+        let redis_store = match &global_config.redis {
+            Some(conf) => match crate::storage::redis::RedisStore::new(conf).await {
+                Ok(store) => {
+                    store.spawn_error_watcher();
+                    Some(store)
+                }
+                Err(err) => {
+                    crate::logger::error!(
+                        ?err,
+                        "Failed to initialize Redis; continuing without it"
+                    );
+                    None
+                }
+            },
+            None => None,
+        };
+
         let tenants_app_state = {
             #[cfg(feature = "key_custodian")]
             {
@@ -55,10 +76,15 @@ impl GlobalAppState {
                     let tenant_config =
                         TenantConfig::from_global_config(&global_config, tenant_id.clone());
                     #[allow(clippy::expect_used)]
-                    let tenant_app_state =
-                        TenantAppState::new(&global_config, tenant_config, api_client.clone())
-                            .await
-                            .expect("Failed while configuring AppState for tenants");
+                    let tenant_app_state = TenantAppState::new(
+                        &global_config,
+                        tenant_config,
+                        api_client.clone(),
+                        #[cfg(feature = "redis")]
+                        redis_store.as_ref(),
+                    )
+                    .await
+                    .expect("Failed while configuring AppState for tenants");
                     tenants_app_state.insert(tenant_id.clone(), Arc::new(tenant_app_state));
                 }
                 tenants_app_state
@@ -72,6 +98,8 @@ impl GlobalAppState {
             api_client: api_client.clone(),
             known_tenants: HashSet::<String>::from_iter(known_tenants),
             global_config,
+            #[cfg(feature = "redis")]
+            redis_store,
         })
     }
 

Original file line number	Diff line number	Diff line change
`@@ -118,6 +118,8 @@ pub async fn decrypt(`
`118`	`118`	`&global_app_state.global_config,`
`119`	`119`	`tenant_config,`
`120`	`120`	`global_app_state.api_client.clone(),`
	`121`	`+ #[cfg(feature = "redis")]`
	`122`	`+ global_app_state.redis_store.as_ref(),`
`121`	`123`	`)`
`122`	`124`	`.await`
`123`	`125`	`.change_context(error::ApiError::TenantError(`