Editoast improve valkey connection

editoast/Cargo.toml

@@ -118,6 +118,7 @@ deadpool = { version = "0.12.2", features = [
   "rt_tokio_1",
   "unmanaged",
 ] }
+deadpool-redis = "0.18.0"
 derivative.workspace = true
 diesel.workspace = true
 diesel-async = { workspace = true }
@@ -168,12 +169,6 @@ pathfinding = "4.14.0"
 postgis_diesel.workspace = true
 rand.workspace = true
 rangemap.workspace = true
-redis = { version = "0.28", default-features = false, features = [
-  "cluster-async",
-  "connection-manager",
-  "tokio-comp",
-  "tokio-native-tls-comp",
-] }
 regex.workspace = true
 reqwest.workspace = true
 serde = { workspace = true, features = ["derive"] }

editoast/src/client/valkey_config.rs

@@ -11,9 +11,6 @@ pub struct ValkeyConfig {
     #[derivative(Default(value = "false"))]
     #[clap(long, env, default_value_t = false)]
     pub no_cache: bool,
-    #[derivative(Default(value = "false"))]
-    #[clap(long, env, default_value_t = false)]
-    pub is_cluster_client: bool,
     #[derivative(Default(value = r#"Url::parse("redis://localhost:6379").unwrap()"#))]
     #[arg(long, env, default_value_t = Url::parse("redis://localhost:6379").unwrap())]
     /// Valkey url like `redis://[:PASSWORD@]HOST[:PORT][/DATABASE]`
@@ -24,13 +21,11 @@ impl From<ValkeyConfig> for valkey_utils::ValkeyConfig {
     fn from(
         ValkeyConfig {
             no_cache,
-            is_cluster_client,
             valkey_url,
         }: ValkeyConfig,
     ) -> Self {
         valkey_utils::ValkeyConfig {
             no_cache,
-            is_cluster_client,
             valkey_url,
         }
     }

editoast/src/error.rs

@@ -2,11 +2,12 @@ use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use axum::Json;
 use colored::Colorize;
+use deadpool_redis::redis::RedisError;
+use deadpool_redis::PoolError;
 use diesel::result::Error as DieselError;
 use editoast_models::db_connection_pool::DatabasePoolBuildError;
 use editoast_models::db_connection_pool::DatabasePoolError;
 use editoast_models::DatabaseError;
-use redis::RedisError;
 use serde::Deserialize;
 use serde::Serialize;
 use serde_json::{json, Value};
@@ -171,7 +172,16 @@ impl EditoastError for RedisError {
     }
 }
 
-// Handle all json errors
+/// Handle all valkey pool errors
+impl EditoastError for PoolError {
+    fn get_status(&self) -> StatusCode {
+        StatusCode::INTERNAL_SERVER_ERROR
+    }
+
+    fn get_type(&self) -> &str {
+        "editoast:ValkeyPoolError"
+    }
+}
 
 /// Handle all json errors
 impl EditoastError for ValidationErrors {

editoast/src/map/mod.rs

@@ -1,10 +1,10 @@
 mod layer_cache;
 mod layers;
 
+use deadpool_redis::redis::AsyncCommands;
 pub use layers::Layer;
 pub use layers::MapLayers;
 pub use layers::View;
-use redis::AsyncCommands;
 
 pub use self::layer_cache::get_cache_tile_key;
 pub use self::layer_cache::get_layer_cache_prefix;

editoast/src/valkey_utils.rs

@@ -1,19 +1,23 @@
 use std::fmt::Debug;
 
+use deadpool_redis::redis::aio::ConnectionLike;
+use deadpool_redis::redis::cmd;
+use deadpool_redis::redis::Arg;
+use deadpool_redis::redis::AsyncCommands;
+use deadpool_redis::redis::Cmd;
+use deadpool_redis::redis::ErrorKind;
+use deadpool_redis::redis::Pipeline;
+use deadpool_redis::redis::RedisError;
+use deadpool_redis::redis::RedisFuture;
+use deadpool_redis::redis::ToRedisArgs;
+use deadpool_redis::redis::Value;
+use deadpool_redis::Config;
+use deadpool_redis::Connection;
+use deadpool_redis::Pool;
+use deadpool_redis::PoolError;
+use deadpool_redis::Runtime;
 use futures::future;
 use futures::FutureExt;
-use redis::aio::ConnectionLike;
-use redis::aio::ConnectionManager;
-use redis::cluster::ClusterClient;
-use redis::cluster_async::ClusterConnection;
-use redis::cmd;
-use redis::AsyncCommands;
-use redis::Client;
-use redis::ErrorKind;
-use redis::RedisError;
-use redis::RedisFuture;
-use redis::RedisResult;
-use redis::ToRedisArgs;
 use serde::de::DeserializeOwned;
 use serde::Serialize;
 use tracing::{debug, span, trace, Level};
@@ -22,59 +26,54 @@ use url::Url;
 use crate::error::Result;
 
 pub enum ValkeyConnection {
-    Cluster(ClusterConnection),
-    Tokio(ConnectionManager),
+    Tokio(Connection),
     NoCache,
 }
 
-fn no_cache_cmd_handler(cmd: &redis::Cmd) -> std::result::Result<redis::Value, RedisError> {
-    let cmd_name = cmd.args_iter().next().ok_or((
-        redis::ErrorKind::ClientError,
-        "missing a command instruction",
-    ))?;
+fn no_cache_cmd_handler(cmd: &Cmd) -> std::result::Result<Value, RedisError> {
+    let cmd_name = cmd
+        .args_iter()
+        .next()
+        .ok_or((ErrorKind::ClientError, "missing a command instruction"))?;
     let nb_keys = cmd.args_iter().skip(1).count();
     match cmd_name {
-        redis::Arg::Simple(cmd_name_bytes)
+        Arg::Simple(cmd_name_bytes)
             if cmd_name_bytes == "MGET".as_bytes()
                 || cmd_name_bytes == "MSET".as_bytes()
                 || nb_keys > 1 =>
         {
-            Ok(redis::Value::Array(vec![redis::Value::Nil; nb_keys]))
+            Ok(Value::Array(vec![Value::Nil; nb_keys]))
         },
-        redis::Arg::Simple(_)
+        Arg::Simple(_)
             if nb_keys == 1 =>
         {
-            Ok(redis::Value::Nil)
+            Ok(Value::Nil)
         },
-        redis::Arg::Simple(cmd_name_bytes) if cmd_name_bytes == "PING".as_bytes() => Ok(redis::Value::SimpleString("PONG".to_string())),
-        redis::Arg::Simple(cmd_name_bytes) => unimplemented!(
+        Arg::Simple(cmd_name_bytes) if cmd_name_bytes == "PING".as_bytes() => Ok(Value::SimpleString("PONG".to_string())),
+        Arg::Simple(cmd_name_bytes) => unimplemented!(
             "valkey command '{}' is not supported by editoast::valkey_utils::ValkeyConnection with '--no-cache'", String::from_utf8(cmd_name_bytes.to_vec())?
         ),
-        redis::Arg::Cursor => unimplemented!(
+        Arg::Cursor => unimplemented!(
             "valkey cursor mode is not supported by editoast::valkey_utils::ValkeyConnection with '--no-cache'"
         ),
     }
 }
 
 impl ConnectionLike for ValkeyConnection {
-    fn req_packed_command<'a>(&'a mut self, cmd: &'a redis::Cmd) -> RedisFuture<'a, redis::Value> {
+    fn req_packed_command<'a>(&'a mut self, cmd: &'a Cmd) -> RedisFuture<'a, Value> {
         match self {
-            ValkeyConnection::Cluster(connection) => connection.req_packed_command(cmd),
             ValkeyConnection::Tokio(connection) => connection.req_packed_command(cmd),
             ValkeyConnection::NoCache => future::ready(no_cache_cmd_handler(cmd)).boxed(),
         }
     }
 
     fn req_packed_commands<'a>(
         &'a mut self,
-        cmd: &'a redis::Pipeline,
+        cmd: &'a Pipeline,
         offset: usize,
         count: usize,
-    ) -> RedisFuture<'a, Vec<redis::Value>> {
+    ) -> RedisFuture<'a, Vec<Value>> {
         match self {
-            ValkeyConnection::Cluster(connection) => {
-                connection.req_packed_commands(cmd, offset, count)
-            }
             ValkeyConnection::Tokio(connection) => {
                 connection.req_packed_commands(cmd, offset, count)
             }
@@ -84,15 +83,14 @@ impl ConnectionLike for ValkeyConnection {
                     .skip(offset)
                     .take(count)
                     .map(no_cache_cmd_handler)
-                    .collect::<std::result::Result<_, redis::RedisError>>();
+                    .collect::<std::result::Result<_, RedisError>>();
                 future::ready(responses).boxed()
             }
         }
     }
 
     fn get_db(&self) -> i64 {
         match self {
-            ValkeyConnection::Cluster(connection) => connection.get_db(),
             ValkeyConnection::Tokio(connection) => connection.get_db(),
             ValkeyConnection::NoCache => 0,
         }
@@ -267,8 +265,7 @@ impl ValkeyConnection {
 
 #[derive(Clone)]
 pub enum ValkeyClient {
-    Cluster(ClusterClient),
-    Tokio(Client),
+    Tokio(Pool),
     /// This doesn't cache anything. It has no backend.
     NoCache,
 }
@@ -277,7 +274,6 @@ pub enum ValkeyClient {
 pub struct ValkeyConfig {
     /// Disables caching. This should not be used in production.
     pub no_cache: bool,
-    pub is_cluster_client: bool,
     pub valkey_url: Url,
 }
 
@@ -286,30 +282,22 @@ impl ValkeyClient {
         if valkey_config.no_cache {
             return Ok(ValkeyClient::NoCache);
         }
-        if valkey_config.is_cluster_client {
-            return Ok(ValkeyClient::Cluster(
-                redis::cluster::ClusterClient::new(vec![valkey_config.valkey_url]).unwrap(),
-            ));
-        }
         Ok(ValkeyClient::Tokio(
-            redis::Client::open(valkey_config.valkey_url).unwrap(),
+            Config::from_url(valkey_config.valkey_url)
+                .create_pool(Some(Runtime::Tokio1))
+                .unwrap(),
         ))
     }
 
-    pub async fn get_connection(&self) -> RedisResult<ValkeyConnection> {
+    pub async fn get_connection(&self) -> std::result::Result<ValkeyConnection, PoolError> {
         match self {
-            ValkeyClient::Cluster(client) => Ok(ValkeyConnection::Cluster(
-                client.get_async_connection().await?,
-            )),
-            ValkeyClient::Tokio(client) => Ok(ValkeyConnection::Tokio(
-                client.get_connection_manager().await?,
-            )),
+            ValkeyClient::Tokio(pool) => Ok(ValkeyConnection::Tokio(pool.get().await?)),
             ValkeyClient::NoCache => Ok(ValkeyConnection::NoCache),
         }
     }
 
     #[tracing::instrument(skip_all)]
-    pub async fn ping_valkey(&self) -> RedisResult<()> {
+    pub async fn ping_valkey(&self) -> anyhow::Result<()> {
         let mut conn = self.get_connection().await?;
         cmd("PING").query_async::<()>(&mut conn).await?;
         trace!("Valkey ping successful");

editoast/src/views/layers.rs

@@ -7,9 +7,9 @@ use axum::extract::State;
 use axum::http::header::CONTENT_TYPE;
 use axum::response::IntoResponse;
 use axum::Extension;
+use deadpool_redis::redis::AsyncCommands;
 use editoast_authz::BuiltinRole;
 use editoast_derive::EditoastError;
-use redis::AsyncCommands;
 use serde::Deserialize;
 use serde::Serialize;
 use thiserror::Error;

editoast/src/views/mod.rs

@@ -266,7 +266,7 @@ pub enum AppHealthError {
     #[error(transparent)]
     Database(#[from] editoast_models::db_connection_pool::PingError),
     #[error(transparent)]
-    Valkey(#[from] redis::RedisError),
+    Valkey(#[from] anyhow::Error),
     #[error(transparent)]
     Core(#[from] CoreError),
 }

editoast/src/views/test_app.rs

@@ -161,7 +161,6 @@ impl TestAppBuilder {
             },
             valkey_config: ValkeyConfig {
                 no_cache: false,
-                is_cluster_client: false,
                 valkey_url: Url::parse("redis://localhost:6379").unwrap(),
             },
         };