feat(redis): add configuration to set the timeout (#3817)

It adds a configuration to set another timeout than the default one
(2ms) for redis requests. It also change the default timeout to 2ms
(previously set to 1ms)

Example for APQ:
```yaml
apq:
  router:
    cache:
      redis:
        urls: ["redis://..."] 
        timeout: 5ms
```

Fixes #3621

<!-- start metadata -->
---

**Checklist**

Complete the checklist (and note appropriate exceptions) before the PR
is marked ready-for-review.

- [x] Changes are compatible[^1]
- [ ] Documentation[^2] completed
- [ ] Performance impact assessed and acceptable
- Tests added and passing[^3]
    - [ ] Unit Tests
    - [ ] Integration Tests
    - [x] Manual Tests

**Exceptions**

*Note any exceptions here*

**Notes**

[^1]: It may be appropriate to bring upcoming changes to the attention
of other (impacted) groups. Please endeavour to do this before seeking
PR approval. The mechanism for doing this will vary considerably, so use
your judgement as to how and when to do this.
[^2]: Configuration is an important part of many changes. Where
applicable please try to document configuration examples.
[^3]: Tick whichever testing boxes are applicable. If you are adding
Manual Tests, please document the manual testing (extensively) in the
Exceptions.

---------

Signed-off-by: Benjamin Coenen <[email protected]>

.changesets/feat_bnjjj_fix_3621.md

@@ -0,0 +1,15 @@
+### feat(redis): add configuration to set the timeout ([Issue #3621](https://github.com/apollographql/router/issues/3621))
+
+It adds a configuration to set another timeout than the default one (2ms) for redis requests. It also change the default timeout to 2ms (previously set to 1ms)
+
+Example for APQ:
+```yaml
+apq:
+  router:
+    cache:
+      redis:
+        urls: ["redis://..."] 
+        timeout: 5ms
+```
+
+By [@bnjjj](https://github.com/bnjjj) in https://github.com/apollographql/router/pull/3817

apollo-router/src/cache/mod.rs

@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 use std::num::NonZeroUsize;
 use std::sync::Arc;
+use std::time::Duration;
 
 use tokio::sync::broadcast;
 use tokio::sync::oneshot;
@@ -34,11 +35,12 @@ where
     pub(crate) async fn with_capacity(
         capacity: NonZeroUsize,
         redis_urls: Option<Vec<url::Url>>,
+        timeout: Option<Duration>,
         caller: &str,
     ) -> Self {
         Self {
             wait_map: Arc::new(Mutex::new(HashMap::new())),
-            storage: CacheStorage::new(capacity, redis_urls, caller).await,
+            storage: CacheStorage::new(capacity, redis_urls, timeout, caller).await,
         }
     }
 
@@ -49,6 +51,7 @@ where
         Self::with_capacity(
             config.in_memory.limit,
             config.redis.as_ref().map(|c| c.urls.clone()),
+            config.redis.as_ref().and_then(|r| r.timeout),
             caller,
         )
         .await
@@ -211,7 +214,8 @@ mod tests {
     async fn example_cache_usage() {
         let k = "key".to_string();
         let cache =
-            DeduplicatingCache::with_capacity(NonZeroUsize::new(1).unwrap(), None, "test").await;
+            DeduplicatingCache::with_capacity(NonZeroUsize::new(1).unwrap(), None, None, "test")
+                .await;
 
         let entry = cache.get(&k).await;
 
@@ -228,7 +232,8 @@ mod tests {
     #[test(tokio::test)]
     async fn it_should_enforce_cache_limits() {
         let cache: DeduplicatingCache<usize, usize> =
-            DeduplicatingCache::with_capacity(NonZeroUsize::new(13).unwrap(), None, "test").await;
+            DeduplicatingCache::with_capacity(NonZeroUsize::new(13).unwrap(), None, None, "test")
+                .await;
 
         for i in 0..14 {
             let entry = cache.get(&i).await;
@@ -251,7 +256,8 @@ mod tests {
         mock.expect_retrieve().times(1).return_const(1usize);
 
         let cache: DeduplicatingCache<usize, usize> =
-            DeduplicatingCache::with_capacity(NonZeroUsize::new(10).unwrap(), None, "test").await;
+            DeduplicatingCache::with_capacity(NonZeroUsize::new(10).unwrap(), None, None, "test")
+                .await;
 
         // Let's trigger 100 concurrent gets of the same value and ensure only
         // one delegated retrieve is made

apollo-router/src/cache/redis.rs

@@ -115,14 +115,18 @@ where
 }
 
 impl RedisCacheStorage {
-    pub(crate) async fn new(urls: Vec<Url>, ttl: Option<Duration>) -> Result<Self, RedisError> {
+    pub(crate) async fn new(
+        urls: Vec<Url>,
+        ttl: Option<Duration>,
+        timeout: Option<Duration>,
+    ) -> Result<Self, RedisError> {
         let url = Self::preprocess_urls(urls)?;
         let config = RedisConfig::from_url(url.as_str())?;
 
         let client = RedisClient::new(
             config,
             Some(PerformanceConfig {
-                default_command_timeout_ms: 1,
+                default_command_timeout_ms: timeout.map(|t| t.as_millis() as u64).unwrap_or(2),
                 ..Default::default()
             }),
             Some(ReconnectPolicy::new_exponential(0, 1, 2000, 5)),

apollo-router/src/cache/storage.rs

@@ -3,6 +3,7 @@ use std::fmt::{self};
 use std::hash::Hash;
 use std::num::NonZeroUsize;
 use std::sync::Arc;
+use std::time::Duration;
 
 use lru::LruCache;
 use serde::de::DeserializeOwned;
@@ -57,14 +58,15 @@ where
 {
     pub(crate) async fn new(
         max_capacity: NonZeroUsize,
-        _redis_urls: Option<Vec<url::Url>>,
+        redis_urls: Option<Vec<url::Url>>,
+        timeout: Option<Duration>,
         caller: &str,
     ) -> Self {
         Self {
             caller: caller.to_string(),
             inner: Arc::new(Mutex::new(LruCache::new(max_capacity))),
-            redis: if let Some(urls) = _redis_urls {
-                match RedisCacheStorage::new(urls, None).await {
+            redis: if let Some(urls) = redis_urls {
+                match RedisCacheStorage::new(urls, None, timeout).await {
                     Err(e) => {
                         tracing::error!(
                             "could not open connection to Redis for {} caching: {:?}",

apollo-router/src/configuration/mod.rs

@@ -20,7 +20,6 @@ use std::net::SocketAddr;
 use std::num::NonZeroUsize;
 use std::str::FromStr;
 use std::sync::Arc;
-#[cfg(not(test))]
 use std::time::Duration;
 
 use derivative::Derivative;
@@ -866,6 +865,11 @@ impl Default for InMemoryCache {
 pub(crate) struct RedisCache {
     /// List of URLs to the Redis cluster
     pub(crate) urls: Vec<url::Url>,
+
+    #[serde(deserialize_with = "humantime_serde::deserialize", default)]
+    #[schemars(with = "Option<String>", default)]
+    /// Redis request timeout (default: 2ms)
+    pub(crate) timeout: Option<Duration>,
 }
 
 /// TLS related configuration options.

apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap

@@ -83,6 +83,12 @@ expression: "&schema"
                     "urls"
                   ],
                   "properties": {
+                    "timeout": {
+                      "description": "Redis request timeout (default: 2ms)",
+                      "default": null,
+                      "type": "string",
+                      "nullable": true
+                    },
                     "urls": {
                       "description": "List of URLs to the Redis cluster",
                       "type": "array",
@@ -1991,6 +1997,12 @@ expression: "&schema"
                     "urls"
                   ],
                   "properties": {
+                    "timeout": {
+                      "description": "Redis request timeout (default: 2ms)",
+                      "default": null,
+                      "type": "string",
+                      "nullable": true
+                    },
                     "urls": {
                       "description": "List of URLs to the Redis cluster",
                       "type": "array",
@@ -5592,6 +5604,12 @@ expression: "&schema"
             "urls"
           ],
           "properties": {
+            "timeout": {
+              "description": "Redis request timeout (default: 2ms)",
+              "default": null,
+              "type": "string",
+              "nullable": true
+            },
             "urls": {
               "description": "List of URLs to the Redis cluster",
               "type": "array",

apollo-router/src/introspection.rs

@@ -25,7 +25,7 @@ impl Introspection {
         capacity: NonZeroUsize,
     ) -> Self {
         Self {
-            cache: CacheStorage::new(capacity, None, "introspection").await,
+            cache: CacheStorage::new(capacity, None, None, "introspection").await,
             planner,
         }
     }

apollo-router/src/plugins/traffic_shaping/mod.rs

@@ -280,7 +280,17 @@ impl Plugin for TrafficShaping {
                 .as_ref()
                 .map(|cache| cache.urls.clone())
             {
-                Some(RedisCacheStorage::new(urls, None).await?)
+                Some(
+                    RedisCacheStorage::new(
+                        urls,
+                        None,
+                        init.config
+                            .experimental_cache
+                            .as_ref()
+                            .and_then(|c| c.timeout),
+                    )
+                    .await?,
+                )
             } else {
                 None
             };

docs/source/configuration/distributed-caching.mdx

@@ -88,6 +88,7 @@ supergraph:
     experimental_cache:
       redis: #highlight-line
         urls: ["redis://..."] #highlight-line
+        timeout: 5ms # Optional, by default: 2ms
 ```
 
 The value of `urls` is a list of URLs for all Redis instances in your cluster.
@@ -104,6 +105,7 @@ apq:
     cache:
       redis: #highlight-line
         urls: ["redis://..."] #highlight-line
+        timeout: 5ms # Optional, by default: 2ms
 ```
 
 The value of `urls` is a list of URLs for all Redis instances in your cluster.