fix: don't use replicas against standalone Redis instance (#8656)

Author: carodewig

apollo-router/src/cache/redis.rs

@@ -9,7 +9,6 @@ use std::time::Duration;
 
 use fred::clients::Client;
 use fred::clients::Pipeline;
-use fred::clients::Replicas;
 use fred::interfaces::EventInterface;
 #[cfg(test)]
 use fred::mocks::Mocks;
@@ -589,10 +588,6 @@ impl RedisCacheStorage {
         self.inner.next().clone()
     }
 
-    fn replica_client(&self) -> Replicas<Client> {
-        self.client().replicas()
-    }
-
     pub(crate) fn pipeline(&self) -> Pipeline<Client> {
         self.inner.next().pipeline()
     }
@@ -622,26 +617,28 @@ impl RedisCacheStorage {
         options: Options,
     ) -> Result<RedisValue<V>, RedisError> {
         let key = self.make_key(key);
-        match self.ttl {
-            Some(ttl) if self.reset_ttl => {
-                let pipeline = self.pipeline().with_options(&options);
-                let _: () = pipeline
-                    .get(&key)
-                    .await
-                    .inspect_err(|e| self.record_error(e))?;
-                let _: () = pipeline
-                    .expire(&key, ttl.as_secs() as i64, None)
-                    .await
-                    .inspect_err(|e| self.record_error(e))?;
-
-                let (value, _timeout_set): (RedisValue<V>, bool) =
-                    pipeline.all().await.inspect_err(|e| self.record_error(e))?;
-                Ok(value)
-            }
-            _ => {
-                let client = self.replica_client().with_options(&options);
-                client.get(key).await.inspect_err(|e| self.record_error(e))
-            }
+        if self.reset_ttl
+            && let Some(ttl) = self.ttl
+        {
+            let pipeline = self.pipeline().with_options(&options);
+            let _: () = pipeline
+                .get(&key)
+                .await
+                .inspect_err(|e| self.record_error(e))?;
+            let _: () = pipeline
+                .expire(&key, ttl.as_secs() as i64, None)
+                .await
+                .inspect_err(|e| self.record_error(e))?;
+
+            let (value, _timeout_set): (RedisValue<V>, bool) =
+                pipeline.all().await.inspect_err(|e| self.record_error(e))?;
+            Ok(value)
+        } else if self.is_cluster {
+            let client = self.client().replicas().with_options(&options);
+            client.get(key).await.inspect_err(|e| self.record_error(e))
+        } else {
+            let client = self.client().with_options(&options);
+            client.get(key).await.inspect_err(|e| self.record_error(e))
         }
     }
 
@@ -664,11 +661,12 @@ impl RedisCacheStorage {
         //    - https://redis.io/docs/latest/commands/mget/
 
         tracing::trace!("getting multiple values from redis: {:?}", keys);
-        let client = self.replica_client().with_options(&options);
+        let client = self.client();
 
         if keys.len() == 1 {
             let key = self.make_key(keys.swap_remove(0));
             let res = client
+                .with_options(&options)
                 .get(key)
                 .await
                 .inspect_err(|e| self.record_error(e))
@@ -689,6 +687,8 @@ impl RedisCacheStorage {
             }
 
             // then we query all the key groups at the same time
+            // use `client.replicas()` since we're in a cluster and can take advantage of read-replicas
+            let client = client.replicas().with_options(&options);
             let mut tasks = Vec::new();
             for (_shard, (indexes, keys)) in h {
                 let client = client.clone();
@@ -722,6 +722,7 @@ impl RedisCacheStorage {
                 .map(|k| self.make_key(k))
                 .collect::<Vec<_>>();
             client
+                .with_options(&options)
                 .mget(keys)
                 .await
                 .inspect_err(|e| self.record_error(e))

apollo-router/src/plugins/response_cache/storage/mod.rs

@@ -8,7 +8,7 @@ use std::time::Duration;
 use std::time::Instant;
 
 pub(super) use error::Error;
-use tokio_util::future::FutureExt;
+use tokio_util::time::FutureExt;
 
 use super::cache_control::CacheControl;
 use crate::plugins::response_cache::invalidation::InvalidationKind;

apollo-router/src/plugins/response_cache/storage/redis.rs

@@ -18,7 +18,7 @@ use serde::Serialize;
 use tokio::sync::broadcast;
 use tokio::sync::mpsc;
 use tokio::task::JoinSet;
-use tokio_util::future::FutureExt;
+use tokio_util::time::FutureExt;
 use tower::BoxError;
 
 use super::CacheEntry;