Bug: [FIX] add unresponsive timeout for redis



This will check if a connection is unresponsive, if it is it will try to re-construct the connection
diff --git a/Cargo.lock b/Cargo.lock
index 2c9293c1701..b466a161097 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2610,9 +2610,9 @@ dependencies = [
 
 [[package]]
 name = "fred"
-version = "7.1.0"
+version = "7.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9282e65613822eea90c99872c51afa1de61542215cb11f91456a93f50a5a131a"
+checksum = "b99c2b48934cd02a81032dd7428b7ae831a27794275bc94eba367418db8a9e55"
 dependencies = [
  "arc-swap",
  "async-trait",
diff --git a/config/config.example.toml b/config/config.example.toml
index abe3f6b4e08..028f689d796 100644
--- a/config/config.example.toml
+++ b/config/config.example.toml
@@ -54,9 +54,10 @@ use_legacy_version = false        # Resp protocol for fred crate (set this to tr
 stream_read_count = 1             # Default number of entries to read from stream if not provided in stream read options
 auto_pipeline = true              # Whether or not the client should automatically pipeline commands across tasks when possible.
 disable_auto_backpressure = false # Whether or not to disable the automatic backpressure features when pipelining is enabled.
-max_in_flight_commands = 5000     # The maximum number of in-flight commands (per connection) before backpressure will be applied.
-default_command_timeout = 0       # An optional timeout to apply to all commands.
-max_feed_count = 200              # The maximum number of frames that will be fed to a socket before flushing.
+max_in_flight_commands = 5000 # The maximum number of in-flight commands (per connection) before backpressure will be applied.
+default_command_timeout = 30  # An optional timeout to apply to all commands. In seconds
+unresponsive_timeout = 10 # An optional timeout for Unresponsive commands in seconds. This should be less than default_command_timeout.
+max_feed_count = 200       # The maximum number of frames that will be fed to a socket before flushing.
 
 # This section provides configs for currency conversion api
 [forex_api]
diff --git a/config/deployments/env_specific.toml b/config/deployments/env_specific.toml
index c8c118d9827..990796c79bc 100644
--- a/config/deployments/env_specific.toml
+++ b/config/deployments/env_specific.toml
@@ -169,9 +169,10 @@ use_legacy_version = false                                              # RESP p
 stream_read_count = 1                                                   # Default number of entries to read from stream if not provided in stream read options
 auto_pipeline = true                                                    # Whether or not the client should automatically pipeline commands across tasks when possible.
 disable_auto_backpressure = false                                       # Whether or not to disable the automatic backpressure features when pipelining is enabled.
-max_in_flight_commands = 5000                                           # The maximum number of in-flight commands (per connection) before backpressure will be applied.
-default_command_timeout = 0                                             # An optional timeout to apply to all commands.
-max_feed_count = 200                                                    # The maximum number of frames that will be fed to a socket before flushing.
+max_in_flight_commands = 5000 # The maximum number of in-flight commands (per connection) before backpressure will be applied.
+default_command_timeout = 30  # An optional timeout to apply to all commands. In seconds
+unresponsive_timeout = 10 # An optional timeout for Unresponsive commands in seconds. This should be less than default_command_timeout.
+max_feed_count = 200       # The maximum number of frames that will be fed to a socket before flushing.
 cluster_enabled = true                                                  # boolean
 cluster_urls = ["redis.cluster.uri-1:8080", "redis.cluster.uri-2:4115"] # List of redis cluster urls
 
diff --git a/config/development.toml b/config/development.toml
index d69719bd419..2fab7deeb3e 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -44,7 +44,8 @@ stream_read_count = 1
 auto_pipeline = true
 disable_auto_backpressure = false
 max_in_flight_commands = 5000
-default_command_timeout = 0
+default_command_timeout = 30
+unresponsive_timeout = 10
 max_feed_count = 200
 
 
diff --git a/config/docker_compose.toml b/config/docker_compose.toml
index 8170132bb85..e0de31dfbb7 100644
--- a/config/docker_compose.toml
+++ b/config/docker_compose.toml
@@ -78,7 +78,8 @@ stream_read_count = 1
 auto_pipeline = true
 disable_auto_backpressure = false
 max_in_flight_commands = 5000
-default_command_timeout = 0
+default_command_timeout = 30
+unresponsive_timeout = 10
 max_feed_count = 200
 
 [cors]
diff --git a/crates/redis_interface/Cargo.toml b/crates/redis_interface/Cargo.toml
index 32e850a073f..85cfef3df54 100644
--- a/crates/redis_interface/Cargo.toml
+++ b/crates/redis_interface/Cargo.toml
@@ -9,7 +9,7 @@ license.workspace = true
 
 [dependencies]
 error-stack = "0.3.1"
-fred = { version = "7.0.0", features = ["metrics", "partial-tracing", "subscriber-client"] }
+fred = { version = "7.1.2", features = ["metrics", "partial-tracing", "subscriber-client", "check-unresponsive"] }
 futures = "0.3"
 serde = { version = "1.0.193", features = ["derive"] }
 thiserror = "1.0.40"
diff --git a/crates/redis_interface/src/lib.rs b/crates/redis_interface/src/lib.rs
index 33d40ebe155..b46e4aec191 100644
--- a/crates/redis_interface/src/lib.rs
+++ b/crates/redis_interface/src/lib.rs
@@ -132,6 +132,11 @@ impl RedisConnectionPool {
             },
         };
 
+        let connection_config = fred::types::ConnectionConfig {
+            unresponsive_timeout: std::time::Duration::from_secs(conf.unresponsive_timeout),
+            ..fred::types::ConnectionConfig::default()
+        };
+
         if !conf.use_legacy_version {
             config.version = fred::types::RespVersion::RESP3;
         }
@@ -151,7 +156,7 @@ impl RedisConnectionPool {
         let pool = fred::prelude::RedisPool::new(
             config,
             Some(perf),
-            None,
+            Some(connection_config),
             Some(reconnect_policy),
             conf.pool_size,
         )
@@ -201,6 +206,15 @@ impl RedisConnectionPool {
             }
         }
     }
+
+    pub async fn on_unresponsive(&self) {
+        let _ = self.pool.clients().iter().map(|client| {
+            client.on_unresponsive(|server| {
+                logger::warn!(redis_server =?server.host, "Redis server is unresponsive");
+                Ok(())
+            })
+        });
+    }
 }
 
 struct RedisConfig {
diff --git a/crates/redis_interface/src/types.rs b/crates/redis_interface/src/types.rs
index 4ebb620c36a..cb883fc1693 100644
--- a/crates/redis_interface/src/types.rs
+++ b/crates/redis_interface/src/types.rs
@@ -57,6 +57,7 @@ pub struct RedisSettings {
     pub max_in_flight_commands: u64,
     pub default_command_timeout: u64,
     pub max_feed_count: u64,
+    pub unresponsive_timeout: u64,
 }
 
 impl RedisSettings {
@@ -76,7 +77,17 @@ impl RedisSettings {
                 "Redis `cluster_urls` must be specified if `cluster_enabled` is `true`".into(),
             ))
             .into_report()
-        })
+        })?;
+
+        when(
+            self.default_command_timeout < self.unresponsive_timeout,
+            || {
+                Err(errors::RedisError::InvalidConfiguration(
+                    "Unresponsive timeout cannot be greater than the command timeout".into(),
+                ))
+                .into_report()
+            },
+        )
     }
 }
 
@@ -97,8 +108,9 @@ impl Default for RedisSettings {
             auto_pipeline: true,
             disable_auto_backpressure: false,
             max_in_flight_commands: 5000,
-            default_command_timeout: 0,
+            default_command_timeout: 30,
             max_feed_count: 200,
+            unresponsive_timeout: 10,
         }
     }
 }

## Type of Change
<!-- Put an `x` in the boxes that apply -->
- [x] Enhancement

## Description
<!-- Describe your changes in detail -->
Enable `check-unresponsive` in fred and update the version to `7.1.2`

## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->
This will retry the command automatically every second(ref: https://github.com/aembke/fred.rs/blob/52bff989263b8bb27030be4e20fd313e65608b44/examples/globals.rs#L22) until `unresponsive_timeout` is exhausted. For retry it tries if one succeeds other will be discarded(ref: https://github.com/aembke/fred.rs/blob/52bff989263b8bb27030be4e20fd313e65608b44/src/router/utils.rs#L566) .

## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
- Sanity Test with Payments and Refunds.

**This change cannot be tested since we cannot reproduce the situation in sandbox**

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code