feat(labrinth): Redis Cluster (#6771)

* chore(labrinth): bump to redis 1.4.1

* feat(labrinth): redis cluster

* chore: cleanup

* feat(labrinth): cache locking

* fix(labrinth): clippy

* chore(labrinth): cleanup env, remove postcard support

* chore(ci): fix test env for labrinth

* chore(labrinth): bump all key versions

* chore(labrinth): improve redis key identities handling

* chore(labrinth): simplify deadline handling

* chore(labrinth): remove unused lease tracking

* chore(labrinth): remove distributed cache locking for now

* chore(labrinth): improve redis backend init error

* feat(labrinth): expose redis read replica strategy

* chore(ci): remove other connection mode tests

* chore: split xredis crate

* feat(xredis): primaries routing

* chore: tombi fmt

* chore: clippy

* chore: update query cache

packages/xredis/src/blocking.rs

@@ -0,0 +1,149 @@
+use std::time::Duration;
+
+use prometheus::Registry;
+
+use super::config::{RedisConfig, RedisTopology};
+use super::connection::RedisBackendBuildError;
+use super::metrics::{
+    LogicalPoolStatus, LogicalPoolStatusProvider,
+    register_blocking_pool_metrics,
+};
+use super::{Error, RedisPool};
+
+const POOL_RETAIN_INTERVAL: Duration = Duration::from_secs(30);
+const MAX_IDLE_CONNECTION_AGE: Duration = Duration::from_secs(5 * 60);
+const MAX_STANDALONE_CONNECTION_AGE: Duration = Duration::from_secs(120);
+
+/// A pool of Redis connections used for blocking operations.
+#[derive(Clone)]
+pub(super) struct RedisBlockingPool {
+    inner: RedisBlockingPoolInner,
+}
+
+#[derive(Clone)]
+enum RedisBlockingPoolInner {
+    Standalone(deadpool_redis::Pool),
+    Cluster(deadpool_redis::cluster::Pool),
+}
+
+impl RedisBlockingPool {
+    pub(super) async fn new(
+        config: &RedisConfig,
+    ) -> Result<Self, RedisBackendBuildError> {
+        let pool_size = config.blocking_pool_size();
+        let inner = match config.topology() {
+            RedisTopology::Standalone => {
+                let connection_config = redis::AsyncConnectionConfig::new()
+                    .set_connection_timeout(None)
+                    .set_response_timeout(None);
+                let manager = deadpool_redis::Manager::new_with_config(
+                    config.seed_urls()[0].clone(),
+                    connection_config,
+                )?;
+                let pool = deadpool_redis::Pool::builder(manager)
+                    .max_size(pool_size.max())
+                    .wait_timeout(Some(Duration::from_millis(
+                        config.wait_timeout_ms(),
+                    )))
+                    .runtime(deadpool_redis::Runtime::Tokio1)
+                    .build()?;
+                retain_standalone_pool(pool.clone());
+                RedisBlockingPoolInner::Standalone(pool)
+            }
+            RedisTopology::Cluster => {
+                let manager = deadpool_redis::cluster::Manager::new(
+                    config.seed_urls().to_vec(),
+                    false,
+                )?;
+                let pool = deadpool_redis::cluster::Pool::builder(manager)
+                    .max_size(pool_size.max())
+                    .wait_timeout(Some(Duration::from_millis(
+                        config.wait_timeout_ms(),
+                    )))
+                    .runtime(deadpool_redis::Runtime::Tokio1)
+                    .build()?;
+                retain_cluster_pool(pool.clone());
+                RedisBlockingPoolInner::Cluster(pool)
+            }
+        };
+
+        Ok(Self { inner })
+    }
+
+    pub(super) fn register_metrics(
+        &self,
+        registry: &Registry,
+    ) -> Result<(), prometheus::Error> {
+        register_blocking_pool_metrics(registry, self.clone())
+    }
+
+    async fn brpop(
+        &self,
+        key: &str,
+        timeout: Duration,
+    ) -> Result<Option<[Vec<u8>; 2]>, Error> {
+        if timeout.is_zero() {
+            return Err(Error::InvalidBlockingTimeout);
+        }
+
+        let mut command = redis::cmd("BRPOP");
+        command.arg(key).arg(timeout.as_secs_f64());
+
+        let response: Option<(Vec<u8>, Vec<u8>)> = match &self.inner {
+            RedisBlockingPoolInner::Standalone(pool) => {
+                command.query_async(&mut pool.get().await?).await?
+            }
+            RedisBlockingPoolInner::Cluster(pool) => {
+                command.query_async(&mut pool.get().await?).await?
+            }
+        };
+
+        Ok(response.map(|(key, value)| [key, value]))
+    }
+}
+
+impl LogicalPoolStatusProvider for RedisBlockingPool {
+    fn logical_pool_status(&self) -> LogicalPoolStatus {
+        match &self.inner {
+            RedisBlockingPoolInner::Standalone(pool) => {
+                LogicalPoolStatus::from_deadpool(pool.status())
+            }
+            RedisBlockingPoolInner::Cluster(pool) => {
+                LogicalPoolStatus::from_deadpool(pool.status())
+            }
+        }
+    }
+}
+
+impl RedisPool {
+    pub async fn brpop(
+        &self,
+        key: &str,
+        timeout: Duration,
+    ) -> Result<Option<[Vec<u8>; 2]>, Error> {
+        self.blocking.brpop(key, timeout).await
+    }
+}
+
+fn retain_standalone_pool(pool: deadpool_redis::Pool) {
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(POOL_RETAIN_INTERVAL).await;
+            pool.retain(|_, metrics| {
+                metrics.last_used() < MAX_IDLE_CONNECTION_AGE
+                    && metrics.created.elapsed() < MAX_STANDALONE_CONNECTION_AGE
+            });
+        }
+    });
+}
+
+fn retain_cluster_pool(pool: deadpool_redis::cluster::Pool) {
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(POOL_RETAIN_INTERVAL).await;
+            pool.retain(|_, metrics| {
+                metrics.last_used() < MAX_IDLE_CONNECTION_AGE
+            });
+        }
+    });
+}

packages/xredis/src/cache.rs

@@ -0,0 +1,732 @@
+use std::borrow::Cow;
+use std::collections::HashMap;
+use std::fmt::{Debug, Display};
+use std::future::Future;
+use std::hash::Hash;
+use std::str::FromStr;
+
+use ariadne::ids::base62_impl::{parse_base62, to_base62};
+use chrono::{TimeZone, Utc};
+use dashmap::DashMap;
+use futures::stream::{FuturesUnordered, StreamExt};
+use redis::aio::ConnectionLike;
+use serde::de::DeserializeOwned;
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+use tokio::time::{Instant, timeout_at};
+use tracing::{Instrument, info_span};
+
+use crate::Error;
+
+use super::commands;
+use super::connection::RoutableConnection;
+use super::key::KeyBuilder;
+
+mod locking;
+
+use locking::{
+    LockAcquisition, LockCoordinator, LockWaiter, WAIT_TIMEOUT, normalize_key,
+};
+
+const FILL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
+
+pub(super) trait ConnectionProvider {
+    type Connection: ConnectionLike + RoutableConnection;
+
+    fn connect(
+        &self,
+    ) -> impl Future<Output = Result<Self::Connection, Error>> + Send;
+}
+
+#[derive(Clone, Copy)]
+enum CacheReadRouting {
+    ReplicaOptional,
+    Primary,
+}
+
+#[repr(u8)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Codec {
+    Raw = 0,
+    Lz4 = 1,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum EncodingFormat {
+    Json,
+}
+
+#[derive(Debug, Error)]
+#[error("invalid redis codec")]
+pub struct InvalidCodec;
+
+#[derive(Debug, Error)]
+#[error("invalid redis encoding format")]
+pub struct InvalidEncodingFormat;
+
+impl TryFrom<u8> for Codec {
+    type Error = InvalidCodec;
+
+    fn try_from(value: u8) -> Result<Self, Self::Error> {
+        match value {
+            0 => Ok(Self::Raw),
+            1 => Ok(Self::Lz4),
+            _ => Err(InvalidCodec),
+        }
+    }
+}
+
+impl FromStr for Codec {
+    type Err = InvalidCodec;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value {
+            "lz4" => Ok(Self::Lz4),
+            _ => Err(InvalidCodec),
+        }
+    }
+}
+
+impl FromStr for EncodingFormat {
+    type Err = InvalidEncodingFormat;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value {
+            "json" => Ok(Self::Json),
+            _ => Err(InvalidEncodingFormat),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct CacheSettings {
+    pub default_expiry: i64,
+    pub actual_expiry: i64,
+    pub version_default_expiry: i64,
+    pub version_actual_expiry: i64,
+    pub encoding_format: EncodingFormat,
+    pub compression_algorithm: Codec,
+    pub compression_level: i32,
+    pub compression_threshold_bytes: usize,
+    pub compression_min_savings_ratio: f64,
+}
+
+impl CacheSettings {
+    pub fn encode_value<T: Serialize>(
+        &self,
+        value: &T,
+    ) -> Result<Vec<u8>, Error> {
+        let mut value = match self.encoding_format {
+            EncodingFormat::Json => serde_json::to_vec(value)?,
+        };
+
+        if self.compression_level > 0
+            && self.compression_algorithm == Codec::Lz4
+            && value.len() >= self.compression_threshold_bytes
+        {
+            let compressed = lz4_flex::block::compress_prepend_size(&value);
+            let savings_ratio = value.len().saturating_sub(compressed.len())
+                as f64
+                / value.len().max(1) as f64
+                * 100.0;
+
+            if savings_ratio >= self.compression_min_savings_ratio {
+                let mut encoded = Vec::with_capacity(compressed.len() + 1);
+                encoded.push(Codec::Lz4 as u8);
+                encoded.extend(compressed);
+                return Ok(encoded);
+            }
+        }
+
+        let mut encoded = Vec::with_capacity(value.len() + 1);
+        encoded.push(Codec::Raw as u8);
+        encoded.append(&mut value);
+        Ok(encoded)
+    }
+
+    pub fn decode_value<T>(&self, value: &[u8]) -> Option<T>
+    where
+        T: for<'a> Deserialize<'a>,
+    {
+        let (codec, value) = value.split_first()?;
+        let value = match Codec::try_from(*codec).ok()? {
+            Codec::Raw => Cow::Borrowed(value),
+            Codec::Lz4 => Cow::Owned(
+                lz4_flex::block::decompress_size_prepended(value).ok()?,
+            ),
+        };
+
+        match self.encoding_format {
+            EncodingFormat::Json => serde_json::from_slice(&value).ok(),
+        }
+    }
+
+    fn expiries(&self, namespace: &str) -> (i64, i64) {
+        match namespace
+            .split_once(':')
+            .map(|value| value.0)
+            .unwrap_or(namespace)
+        {
+            "versions" | "versions_files" => {
+                (self.version_default_expiry, self.version_actual_expiry)
+            }
+            _ => (self.default_expiry, self.actual_expiry),
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct CacheManager {
+    key_builder: KeyBuilder,
+    settings: CacheSettings,
+    locking: LockCoordinator,
+}
+
+impl CacheManager {
+    pub fn new(key_builder: KeyBuilder, settings: CacheSettings) -> Self {
+        Self {
+            locking: LockCoordinator::new(),
+            key_builder,
+            settings,
+        }
+    }
+
+    pub fn settings(&self) -> &CacheSettings {
+        &self.settings
+    }
+
+    #[tracing::instrument(skip(self, provider, keys, closure))]
+    pub async fn get_cached_keys<P, F, Fut, T, K, E>(
+        &self,
+        provider: &P,
+        namespace: &str,
+        keys: &[K],
+        closure: F,
+    ) -> Result<Vec<T>, E>
+    where
+        P: ConnectionProvider,
+        F: FnOnce(Vec<K>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, T>, E>>,
+        E: From<Error>,
+        T: Serialize + DeserializeOwned,
+        K: Display
+            + Hash
+            + Eq
+            + PartialEq
+            + Clone
+            + DeserializeOwned
+            + Serialize
+            + Debug,
+    {
+        Ok(self
+            .get_cached_keys_raw(provider, namespace, keys, closure)
+            .await?
+            .into_values()
+            .collect())
+    }
+
+    #[tracing::instrument(skip(self, provider, keys, closure))]
+    pub async fn get_cached_keys_raw<P, F, Fut, T, K, E>(
+        &self,
+        provider: &P,
+        namespace: &str,
+        keys: &[K],
+        closure: F,
+    ) -> Result<HashMap<K, T>, E>
+    where
+        P: ConnectionProvider,
+        F: FnOnce(Vec<K>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, T>, E>>,
+        E: From<Error>,
+        T: Serialize + DeserializeOwned,
+        K: Display
+            + Hash
+            + Eq
+            + PartialEq
+            + Clone
+            + DeserializeOwned
+            + Serialize
+            + Debug,
+    {
+        self.get_cached_keys_raw_with_slug(
+            provider,
+            namespace,
+            None,
+            false,
+            keys,
+            |ids| async move {
+                Ok(closure(ids)
+                    .await?
+                    .into_iter()
+                    .map(|(key, value)| (key, (None::<String>, value)))
+                    .collect())
+            },
+        )
+        .await
+    }
+
+    #[tracing::instrument(skip(self, provider, keys, closure))]
+    pub async fn get_cached_keys_with_slug<P, F, Fut, T, I, K, S, E>(
+        &self,
+        provider: &P,
+        namespace: &str,
+        slug_namespace: &str,
+        case_sensitive: bool,
+        keys: &[I],
+        closure: F,
+    ) -> Result<Vec<T>, E>
+    where
+        P: ConnectionProvider,
+        F: FnOnce(Vec<I>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, (Option<S>, T)>, E>>,
+        E: From<Error>,
+        T: Serialize + DeserializeOwned,
+        I: Display + Hash + Eq + PartialEq + Clone + Debug,
+        K: Display
+            + Hash
+            + Eq
+            + PartialEq
+            + Clone
+            + DeserializeOwned
+            + Serialize,
+        S: Display + Clone + DeserializeOwned + Serialize + Debug,
+    {
+        Ok(self
+            .get_cached_keys_raw_with_slug(
+                provider,
+                namespace,
+                Some(slug_namespace),
+                case_sensitive,
+                keys,
+                closure,
+            )
+            .await?
+            .into_values()
+            .collect())
+    }
+
+    #[tracing::instrument(skip(self, provider, keys, closure))]
+    pub async fn get_cached_keys_raw_with_slug<P, F, Fut, T, I, K, S, E>(
+        &self,
+        provider: &P,
+        namespace: &str,
+        slug_namespace: Option<&str>,
+        case_sensitive: bool,
+        keys: &[I],
+        closure: F,
+    ) -> Result<HashMap<K, T>, E>
+    where
+        P: ConnectionProvider,
+        F: FnOnce(Vec<I>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, (Option<S>, T)>, E>>,
+        E: From<Error>,
+        T: Serialize + DeserializeOwned,
+        I: Display + Hash + Eq + PartialEq + Clone + Debug,
+        K: Display
+            + Hash
+            + Eq
+            + PartialEq
+            + Clone
+            + DeserializeOwned
+            + Serialize,
+        S: Display + Clone + DeserializeOwned + Serialize + Debug,
+    {
+        let ids = keys
+            .iter()
+            .map(|key| {
+                (normalize_key(&key.to_string(), case_sensitive), key.clone())
+            })
+            .collect::<DashMap<String, I>>();
+
+        if ids.is_empty() {
+            return Ok(HashMap::new());
+        }
+
+        let get_cached_values =
+            |ids: DashMap<String, I>, routing: CacheReadRouting| {
+                async move {
+                    let slug_ids = if let Some(slug_namespace) = slug_namespace
+                    {
+                        async {
+                            let keys = ids
+                                .iter()
+                                .map(|entry| {
+                                    let logical_key = normalize_key(
+                                        &entry.value().to_string(),
+                                        case_sensitive,
+                                    );
+                                    self.key_builder
+                                        .entity(slug_namespace, logical_key)
+                                })
+                                .collect::<Vec<_>>();
+                            let mut connection =
+                                provider.connect().await.map_err(E::from)?;
+                            let values = match routing {
+                                CacheReadRouting::ReplicaOptional => {
+                                    commands::get_many_strings(
+                                        &mut connection,
+                                        &keys,
+                                    )
+                                    .await
+                                }
+                                CacheReadRouting::Primary => {
+                                    commands::get_many_strings_primary(
+                                        &mut connection,
+                                        &keys,
+                                    )
+                                    .await
+                                }
+                            }
+                            .map_err(E::from)?;
+                            Ok::<_, E>(
+                                values
+                                    .into_iter()
+                                    .flatten()
+                                    .collect::<Vec<_>>(),
+                            )
+                        }
+                        .instrument(info_span!("get slug ids"))
+                        .await?
+                    } else {
+                        Vec::new()
+                    };
+
+                    let keys = ids
+                        .iter()
+                        .map(|entry| entry.value().to_string())
+                        .chain(ids.iter().filter_map(|entry| {
+                            parse_base62(&entry.value().to_string())
+                                .ok()
+                                .map(|value| value.to_string())
+                        }))
+                        .chain(slug_ids)
+                        .map(|key| self.key_builder.entity(namespace, key))
+                        .collect::<Vec<_>>();
+
+                    let mut connection =
+                        provider.connect().await.map_err(E::from)?;
+                    let mut cached_values = HashMap::new();
+                    let values = match routing {
+                        CacheReadRouting::ReplicaOptional => {
+                            commands::get_many(&mut connection, &keys).await
+                        }
+                        CacheReadRouting::Primary => {
+                            commands::get_many_primary(&mut connection, &keys)
+                                .await
+                        }
+                    }
+                    .map_err(E::from)?;
+                    for value in values {
+                        if let Some(value) = value.and_then(|value| {
+                            self.settings
+                                .decode_value::<RedisValue<T, K, S>>(&value)
+                        }) {
+                            cached_values.insert(value.key.clone(), value);
+                        }
+                    }
+
+                    Ok::<_, E>((cached_values, ids))
+                }
+                .instrument(info_span!("get_cached_values_closure"))
+            };
+
+        let (default_expiry, actual_expiry) = self.settings.expiries(namespace);
+        let current_time = Utc::now();
+        let mut expired_values = HashMap::new();
+        let mut expired_identities = HashMap::new();
+        let deadline = Instant::now() + WAIT_TIMEOUT;
+
+        let (cached_values_raw, ids) =
+            get_cached_values(ids, CacheReadRouting::ReplicaOptional).await?;
+        let mut cached_values = cached_values_raw
+            .into_iter()
+            .filter_map(|(key, value)| {
+                if Utc.timestamp_opt(value.iat + actual_expiry, 0).unwrap()
+                    < current_time
+                {
+                    let canonical_key = value.key.to_string();
+                    for identity in value_identities(&value, case_sensitive) {
+                        expired_identities
+                            .insert(identity, canonical_key.clone());
+                    }
+                    expired_values.insert(canonical_key, value);
+                    None
+                } else {
+                    remove_resolved_ids(&ids, &value, case_sensitive);
+                    Some((key, value))
+                }
+            })
+            .collect::<HashMap<_, _>>();
+
+        let mut waiters = Vec::new();
+        let mut owned_locks = HashMap::new();
+
+        if !ids.is_empty() {
+            let fetch_ids = ids
+                .iter()
+                .map(|entry| entry.key().clone())
+                .collect::<Vec<_>>();
+            for key in fetch_ids {
+                if !ids.contains_key(&key) {
+                    continue;
+                }
+
+                let lock_key = self.key_builder.entity(namespace, &key);
+                let acquisition = self.locking.acquire(lock_key);
+
+                match acquisition {
+                    LockAcquisition::Owned(guard) => {
+                        owned_locks.insert(key, guard);
+                    }
+                    LockAcquisition::Waiting(waiter) => {
+                        if let Some(canonical_key) =
+                            expired_identities.get(&key).cloned()
+                            && let Some(value) =
+                                expired_values.remove(&canonical_key)
+                        {
+                            remove_resolved_ids(&ids, &value, case_sensitive);
+                            expired_identities.retain(|_, canonical| {
+                                canonical != &canonical_key
+                            });
+                            cached_values.insert(value.key.clone(), value);
+                        } else if let Some((_, raw_key)) = ids.remove(&key) {
+                            waiters.push((raw_key, waiter));
+                        }
+                    }
+                }
+            }
+        }
+
+        let fill_result = if !ids.is_empty() {
+            async {
+                let fetch_ids = ids
+                    .iter()
+                    .map(|entry| entry.value().clone())
+                    .collect::<Vec<_>>();
+
+                let fill_deadline = Instant::now() + FILL_TIMEOUT;
+
+                let values = timeout_at(fill_deadline, closure(fetch_ids))
+                    .await
+                    .map_err(|_| lock_timeout_error(0, waiters.len()))??;
+
+                let mut return_values = HashMap::new();
+                let mut encoded_values = Vec::with_capacity(values.len());
+
+                for (key, (slug, value)) in values {
+                    let value = RedisValue {
+                        key: key.clone(),
+                        iat: Utc::now().timestamp(),
+                        val: value,
+                        alias: slug.clone(),
+                    };
+                    let encoded =
+                        self.settings.encode_value(&value).map_err(E::from)?;
+                    encoded_values.push((key, slug, value, encoded));
+                }
+
+                let mut connection =
+                    provider.connect().await.map_err(E::from)?;
+                for (key, slug, _, encoded) in &encoded_values {
+                    let redis_key =
+                        self.key_builder.entity(namespace, key.to_string());
+                    commands::set(
+                        &mut connection,
+                        &redis_key,
+                        encoded,
+                        default_expiry,
+                    )
+                    .await
+                    .map_err(E::from)?;
+                    if let Some(slug) = slug
+                        && let Some(slug_namespace) = slug_namespace
+                    {
+                        let canonical_key = key.to_string();
+                        let actual_slug =
+                            normalize_key(&slug.to_string(), case_sensitive);
+                        let slug_key = self
+                            .key_builder
+                            .entity(slug_namespace, actual_slug);
+                        commands::set(
+                            &mut connection,
+                            &slug_key,
+                            canonical_key.as_bytes(),
+                            default_expiry,
+                        )
+                        .await
+                        .map_err(E::from)?;
+                    }
+                }
+
+                for (key, _, value, _) in encoded_values {
+                    remove_resolved_ids(&ids, &value, case_sensitive);
+                    return_values.insert(key, value);
+                }
+
+                Result::<_, E>::Ok(return_values)
+            }
+            .await
+        } else {
+            Ok(HashMap::new())
+        };
+
+        drop(owned_locks);
+
+        let operation_result = match fill_result {
+            Ok(mut values) => {
+                if waiters.is_empty() {
+                    Ok(values)
+                } else {
+                    match wait_for_locks(waiters, deadline).await {
+                        Ok(released_ids) => {
+                            let fetch_ids = released_ids
+                                .into_iter()
+                                .map(|key| {
+                                    (
+                                        normalize_key(
+                                            &key.to_string(),
+                                            case_sensitive,
+                                        ),
+                                        key,
+                                    )
+                                })
+                                .collect::<DashMap<_, _>>();
+                            match get_cached_values(
+                                fetch_ids,
+                                CacheReadRouting::Primary,
+                            )
+                            .await
+                            {
+                                Ok((released_values, _)) => {
+                                    values.extend(released_values);
+                                    Ok(values)
+                                }
+                                Err(error) => Err(error),
+                            }
+                        }
+                        Err(error) => Err(E::from(error)),
+                    }
+                }
+            }
+            Err(error) => Err(error),
+        };
+        cached_values.extend(operation_result?);
+
+        Ok(cached_values
+            .into_iter()
+            .map(|(key, value)| (key, value.val))
+            .collect())
+    }
+}
+
+fn remove_resolved_ids<I, T, K, S>(
+    ids: &DashMap<String, I>,
+    value: &RedisValue<T, K, S>,
+    case_sensitive: bool,
+) where
+    K: Display,
+    S: Display,
+{
+    for identity in value_identities(value, case_sensitive) {
+        ids.remove(&normalize_key(&identity, case_sensitive));
+    }
+}
+
+fn value_identities<T, K, S>(
+    value: &RedisValue<T, K, S>,
+    case_sensitive: bool,
+) -> Vec<String>
+where
+    K: Display,
+    S: Display,
+{
+    let mut identities = Vec::with_capacity(5);
+    let canonical_key = value.key.to_string();
+
+    push_identity(&mut identities, canonical_key.clone());
+    if !case_sensitive {
+        push_identity(&mut identities, canonical_key.to_lowercase());
+    }
+
+    if let Ok(decimal_id) = canonical_key.parse::<u64>() {
+        let base62_id = to_base62(decimal_id);
+        push_identity(&mut identities, base62_id.clone());
+
+        if !case_sensitive {
+            push_identity(&mut identities, base62_id.to_lowercase());
+        }
+    } else if let Ok(decimal_id) = parse_base62(&canonical_key) {
+        push_identity(&mut identities, decimal_id.to_string());
+    }
+
+    if let Some(alias) = &value.alias {
+        let alias = alias.to_string();
+        push_identity(&mut identities, alias.clone());
+        if !case_sensitive {
+            push_identity(&mut identities, alias.to_lowercase());
+        }
+    }
+
+    identities
+}
+
+fn push_identity(identities: &mut Vec<String>, identity: String) {
+    if !identities.contains(&identity) {
+        identities.push(identity);
+    }
+}
+
+async fn wait_for_locks<I>(
+    waiters: Vec<(I, LockWaiter)>,
+    deadline: Instant,
+) -> Result<Vec<I>, Error> {
+    let total = waiters.len();
+    let mut released = Vec::with_capacity(total);
+    let mut futures = FuturesUnordered::new();
+    for (key, waiter) in waiters {
+        futures.push(async move {
+            let result = waiter.wait(deadline).await;
+            (key, result)
+        });
+    }
+
+    while let Some((key, result)) = futures.next().await {
+        match result {
+            Ok(()) => {
+                released.push(key);
+            }
+            Err(error)
+                if is_lock_timeout(&error) || Instant::now() >= deadline =>
+            {
+                return Err(lock_timeout_error(released.len(), total));
+            }
+            Err(error) => return Err(error),
+        }
+    }
+    Ok(released)
+}
+
+fn is_lock_timeout(error: &Error) -> bool {
+    matches!(error, Error::LocalCacheTimeout { .. })
+}
+
+fn lock_timeout_error(locks_released: usize, locks_waiting: usize) -> Error {
+    Error::LocalCacheTimeout {
+        released: locks_released,
+        total: locks_waiting,
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct RedisValue<T, K, S> {
+    key: K,
+    alias: Option<S>,
+    iat: i64,
+    val: T,
+}
+
+impl<T, K, S> RedisValue<T, K, S> {
+    pub fn value(&self) -> &T {
+        &self.val
+    }
+}

packages/xredis/src/cache/locking.rs

@@ -0,0 +1,17 @@
+mod local;
+
+use std::time::Duration;
+
+pub(super) use self::local::{LockAcquisition, LockCoordinator, LockWaiter};
+
+pub(super) const WAIT_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Normalize only the requested lookup form's case. Raw IDs and aliases remain
+/// distinct lock identities and may therefore fill concurrently.
+pub(super) fn normalize_key(key: &str, case_sensitive: bool) -> String {
+    if case_sensitive {
+        key.to_owned()
+    } else {
+        key.to_lowercase()
+    }
+}

packages/xredis/src/cache/locking/local.rs

@@ -0,0 +1,121 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use dashmap::DashMap;
+use dashmap::mapref::entry::Entry;
+use tokio::sync::Notify;
+use tokio::time::{Instant, timeout_at};
+
+use crate::Error;
+
+#[derive(Clone)]
+pub(in crate::cache) struct LockCoordinator {
+    locks: Arc<DashMap<String, Arc<LockState>>>,
+}
+
+impl LockCoordinator {
+    pub(in crate::cache) fn new() -> Self {
+        Self {
+            locks: Arc::new(DashMap::with_capacity(2048)),
+        }
+    }
+
+    pub(in crate::cache) fn acquire(&self, key: String) -> LockAcquisition {
+        match self.locks.entry(key.clone()) {
+            Entry::Occupied(entry) => LockAcquisition::Waiting(LockWaiter {
+                state: entry.get().clone(),
+            }),
+            Entry::Vacant(entry) => {
+                let state = Arc::new(LockState::new());
+                entry.insert(state.clone());
+                LockAcquisition::Owned(OwnedLockGuard {
+                    locks: self.locks.clone(),
+                    key,
+                    state,
+                    released: false,
+                })
+            }
+        }
+    }
+}
+
+pub(in crate::cache) enum LockAcquisition {
+    Owned(OwnedLockGuard),
+    Waiting(LockWaiter),
+}
+
+pub(in crate::cache) struct OwnedLockGuard {
+    locks: Arc<DashMap<String, Arc<LockState>>>,
+    key: String,
+    state: Arc<LockState>,
+    released: bool,
+}
+
+impl OwnedLockGuard {
+    fn release_inner(&mut self) {
+        if self.released {
+            return;
+        }
+
+        self.released = true;
+        self.locks
+            .remove_if(&self.key, |_, state| Arc::ptr_eq(state, &self.state));
+        self.state.released.store(true, Ordering::Release);
+        self.state.notify.notify_waiters();
+    }
+}
+
+impl Drop for OwnedLockGuard {
+    fn drop(&mut self) {
+        self.release_inner();
+    }
+}
+
+pub(in crate::cache) struct LockWaiter {
+    state: Arc<LockState>,
+}
+
+impl LockWaiter {
+    pub(in crate::cache) async fn wait(
+        self,
+        deadline: Instant,
+    ) -> Result<(), Error> {
+        loop {
+            if self.state.released.load(Ordering::Acquire) {
+                return Ok(());
+            }
+
+            let notified = self.state.notify.notified();
+            tokio::pin!(notified);
+            notified.as_mut().enable();
+            if self.state.released.load(Ordering::Acquire) {
+                return Ok(());
+            }
+
+            timeout_at(deadline, notified)
+                .await
+                .map_err(|_| lock_timeout())?;
+        }
+    }
+}
+
+struct LockState {
+    released: AtomicBool,
+    notify: Notify,
+}
+
+impl LockState {
+    fn new() -> Self {
+        Self {
+            released: AtomicBool::new(false),
+            notify: Notify::new(),
+        }
+    }
+}
+
+fn lock_timeout() -> Error {
+    Error::LocalCacheTimeout {
+        released: 0,
+        total: 1,
+    }
+}

packages/xredis/src/commands.rs

@@ -0,0 +1,237 @@
+use std::fmt::Debug;
+
+use redis::aio::ConnectionLike;
+use redis::{FromRedisValue, ToRedisArgs};
+
+use crate::Error;
+
+use super::cache::CacheSettings;
+use super::connection::RoutableConnection;
+use super::routing::primary_mget_routing;
+use super::util::cmd;
+
+pub const MGET_CHUNK_SIZE: usize = 32;
+
+#[tracing::instrument(skip_all)]
+pub async fn set<C, D>(
+    connection: &mut C,
+    key: &str,
+    data: D,
+    expiry: i64,
+) -> Result<(), Error>
+where
+    C: ConnectionLike,
+    D: ToRedisArgs + Send + Sync + Debug,
+{
+    cmd("SET")
+        .arg(key)
+        .arg(data)
+        .arg("EX")
+        .arg(expiry)
+        .query_async::<()>(connection)
+        .await?;
+    Ok(())
+}
+
+#[tracing::instrument(skip_all)]
+pub async fn set_serialized<C, D>(
+    connection: &mut C,
+    key: &str,
+    data: D,
+    expiry: Option<i64>,
+    settings: &CacheSettings,
+) -> Result<(), Error>
+where
+    C: ConnectionLike,
+    D: serde::Serialize,
+{
+    set(
+        connection,
+        key,
+        settings.encode_value(&data)?,
+        expiry.unwrap_or(settings.default_expiry),
+    )
+    .await
+}
+
+#[tracing::instrument(skip_all)]
+pub async fn get<C>(
+    connection: &mut C,
+    key: &str,
+) -> Result<Option<String>, Error>
+where
+    C: ConnectionLike,
+{
+    Ok(cmd("GET").arg(key).query_async(connection).await?)
+}
+
+/// Issues ordinary `MGET` commands in bounded chunks. Cluster routing and
+/// result ordering remain redis-rs's responsibility; multiple chunks are not
+/// an atomic snapshot.
+#[tracing::instrument(skip_all)]
+pub async fn get_many<C>(
+    connection: &mut C,
+    keys: &[String],
+) -> Result<Vec<Option<Vec<u8>>>, Error>
+where
+    C: ConnectionLike,
+{
+    get_many_as(connection, keys).await
+}
+
+#[tracing::instrument(skip_all)]
+pub async fn get_many_strings<C>(
+    connection: &mut C,
+    keys: &[String],
+) -> Result<Vec<Option<String>>, Error>
+where
+    C: ConnectionLike,
+{
+    get_many_as(connection, keys).await
+}
+
+pub(super) async fn get_many_primary<C>(
+    connection: &mut C,
+    keys: &[String],
+) -> Result<Vec<Option<Vec<u8>>>, Error>
+where
+    C: RoutableConnection,
+{
+    get_many_primary_as(connection, keys).await
+}
+
+pub(super) async fn get_many_strings_primary<C>(
+    connection: &mut C,
+    keys: &[String],
+) -> Result<Vec<Option<String>>, Error>
+where
+    C: RoutableConnection,
+{
+    get_many_primary_as(connection, keys).await
+}
+
+pub(super) async fn get_many_as<C, T>(
+    connection: &mut C,
+    keys: &[String],
+) -> Result<Vec<Option<T>>, Error>
+where
+    C: ConnectionLike,
+    T: FromRedisValue,
+{
+    let mut values = Vec::with_capacity(keys.len());
+    for chunk in keys.chunks(MGET_CHUNK_SIZE) {
+        let part = cmd("MGET")
+            .arg(chunk)
+            .query_async::<Vec<Option<T>>>(connection)
+            .await?;
+        values.extend(part);
+    }
+    Ok(values)
+}
+
+async fn get_many_primary_as<C, T>(
+    connection: &mut C,
+    keys: &[String],
+) -> Result<Vec<Option<T>>, Error>
+where
+    C: RoutableConnection,
+    T: FromRedisValue,
+{
+    let mut values = Vec::with_capacity(keys.len());
+    for chunk in keys.chunks(MGET_CHUNK_SIZE) {
+        let mut command = redis::cmd("MGET");
+        command.arg(chunk);
+        let value = connection
+            .route_command(command, primary_mget_routing(chunk))
+            .await?;
+        let part =
+            redis::from_redis_value::<Vec<Option<T>>>(value.extract_error()?)
+                .map_err(redis::RedisError::from)?;
+        values.extend(part);
+    }
+    Ok(values)
+}
+
+#[tracing::instrument(skip_all)]
+pub async fn get_deserialized<C, R>(
+    connection: &mut C,
+    key: &str,
+    settings: &CacheSettings,
+) -> Result<Option<R>, Error>
+where
+    C: ConnectionLike,
+    R: for<'a> serde::Deserialize<'a>,
+{
+    let value: Option<Vec<u8>> =
+        cmd("GET").arg(key).query_async(connection).await?;
+    Ok(value.and_then(|value| settings.decode_value(&value)))
+}
+
+#[tracing::instrument(skip_all)]
+pub async fn get_many_deserialized<C, R>(
+    connection: &mut C,
+    keys: &[String],
+    settings: &CacheSettings,
+) -> Result<Vec<Option<R>>, Error>
+where
+    C: ConnectionLike,
+    R: for<'a> serde::Deserialize<'a>,
+{
+    Ok(get_many(connection, keys)
+        .await?
+        .into_iter()
+        .map(|value| value.and_then(|value| settings.decode_value(&value)))
+        .collect())
+}
+
+#[tracing::instrument(skip_all)]
+pub async fn delete<C>(connection: &mut C, key: &str) -> Result<(), Error>
+where
+    C: ConnectionLike,
+{
+    cmd("DEL").arg(key).query_async::<()>(connection).await?;
+    Ok(())
+}
+
+#[tracing::instrument(skip_all)]
+pub async fn delete_many<C>(
+    connection: &mut C,
+    keys: &[String],
+) -> Result<(), Error>
+where
+    C: ConnectionLike,
+{
+    if !keys.is_empty() {
+        cmd("DEL").arg(keys).query_async::<()>(connection).await?;
+    }
+    Ok(())
+}
+
+#[tracing::instrument(skip_all)]
+pub async fn lpush<C, D>(
+    connection: &mut C,
+    key: &str,
+    value: D,
+) -> Result<(), Error>
+where
+    C: ConnectionLike,
+    D: ToRedisArgs + Send + Sync + Debug,
+{
+    cmd("LPUSH")
+        .arg(key)
+        .arg(value)
+        .query_async::<()>(connection)
+        .await?;
+    Ok(())
+}
+
+#[tracing::instrument(skip_all)]
+pub async fn incr<C>(
+    connection: &mut C,
+    key: &str,
+) -> Result<Option<u64>, Error>
+where
+    C: ConnectionLike,
+{
+    Ok(cmd("INCR").arg(key).query_async(connection).await?)
+}

packages/xredis/src/config.rs

@@ -0,0 +1,285 @@
+use std::{fmt, str::FromStr};
+
+use thiserror::Error;
+
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub enum CacheLockingStrategy {
+    #[default]
+    Local,
+    Distributed,
+}
+
+impl CacheLockingStrategy {
+    pub(super) const fn as_str(self) -> &'static str {
+        match self {
+            Self::Local => "local",
+            Self::Distributed => "distributed",
+        }
+    }
+}
+
+impl fmt::Display for CacheLockingStrategy {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(self.as_str())
+    }
+}
+
+#[derive(Debug, Error)]
+#[error("invalid cache locking strategy; expected `local` or `distributed`")]
+pub struct InvalidCacheLockingStrategy;
+
+impl FromStr for CacheLockingStrategy {
+    type Err = InvalidCacheLockingStrategy;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value {
+            "local" => Ok(Self::Local),
+            "distributed" => Ok(Self::Distributed),
+            _ => Err(InvalidCacheLockingStrategy),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RedisTopology {
+    Standalone,
+    Cluster,
+}
+
+#[derive(Debug, Error)]
+#[error("invalid Redis topology; expected `standalone` or `cluster`")]
+pub struct InvalidRedisMode;
+
+impl FromStr for RedisTopology {
+    type Err = InvalidRedisMode;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value {
+            "standalone" => Ok(Self::Standalone),
+            "cluster" => Ok(Self::Cluster),
+            _ => Err(InvalidRedisMode),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RedisConnectionType {
+    Pooled,
+    Multiplexed,
+}
+
+#[derive(Debug, Error)]
+#[error("invalid Redis connection type; expected `pooled` or `multiplexed`")]
+pub struct InvalidRedisConnectionType;
+
+impl FromStr for RedisConnectionType {
+    type Err = InvalidRedisConnectionType;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value {
+            "pooled" => Ok(Self::Pooled),
+            "multiplexed" => Ok(Self::Multiplexed),
+            _ => Err(InvalidRedisConnectionType),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct RedisPoolSize {
+    max: usize,
+    min: usize,
+}
+
+impl RedisPoolSize {
+    fn new(
+        name: &'static str,
+        max: usize,
+        min: usize,
+    ) -> Result<Self, RedisConfigError> {
+        if max == 0 || min > max {
+            return Err(RedisConfigError::InvalidPoolSize { name, max, min });
+        }
+
+        Ok(Self { max, min })
+    }
+
+    pub(crate) fn max(self) -> usize {
+        self.max
+    }
+
+    pub(crate) fn min(self) -> usize {
+        self.min
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub(crate) enum RedisBackendConfig {
+    StandalonePooled(RedisPoolSize),
+    ClusterPooled(RedisPoolSize),
+    ClusterMultiplexed,
+}
+
+#[derive(Debug, Clone)]
+pub struct RedisConfig {
+    mode: RedisTopology,
+    backend: RedisBackendConfig,
+    seed_urls: Vec<String>,
+    wait_timeout_ms: u64,
+    blocking_pool_size: RedisPoolSize,
+    cache_locking_strategy: CacheLockingStrategy,
+    read_replica_strategy: ReadReplicaStrategy,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ReadReplicaStrategy {
+    Primary,
+    RoundRobinReplica,
+    RandomReplica,
+}
+
+#[derive(Debug, Error)]
+#[error(
+    "invalid Redis read replica strategy; expected `primary`, `round_robin`, or `random`"
+)]
+pub struct InvalidRedisReadReplicaStrategy;
+
+impl FromStr for ReadReplicaStrategy {
+    type Err = InvalidRedisReadReplicaStrategy;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value {
+            "primary" => Ok(Self::Primary),
+            "round_robin" => Ok(Self::RoundRobinReplica),
+            "random" => Ok(Self::RandomReplica),
+            _ => Err(InvalidRedisReadReplicaStrategy),
+        }
+    }
+}
+
+#[derive(Debug, Error)]
+pub enum RedisConfigError {
+    #[error("Redis configuration must contain at least one URL")]
+    MissingUrl,
+    #[error("standalone Redis mode requires exactly one URL")]
+    MultipleStandaloneUrls,
+    #[error(
+        "unsupported Redis configuration: `{mode:?}` mode with `{connection_type:?}` connections"
+    )]
+    UnsupportedConnectionType {
+        mode: RedisTopology,
+        connection_type: RedisConnectionType,
+    },
+    #[error(
+        "invalid {name} Redis pool size: minimum {min} must not exceed nonzero maximum {max}"
+    )]
+    InvalidPoolSize {
+        name: &'static str,
+        max: usize,
+        min: usize,
+    },
+    #[error("unsupported Redis cache locking strategy `{strategy}`")]
+    UnsupportedCacheLockingStrategy { strategy: CacheLockingStrategy },
+}
+
+impl RedisConfig {
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        mode: RedisTopology,
+        connection_type: RedisConnectionType,
+        raw_urls: &str,
+        wait_timeout_ms: u64,
+        standalone_pool_size: (usize, usize),
+        cluster_pool_size: (usize, usize),
+        blocking_pool_size: (usize, usize),
+        cache_locking_strategy: CacheLockingStrategy,
+        read_replica_strategy: ReadReplicaStrategy,
+    ) -> Result<Self, RedisConfigError> {
+        if cache_locking_strategy == CacheLockingStrategy::Distributed {
+            return Err(RedisConfigError::UnsupportedCacheLockingStrategy {
+                strategy: cache_locking_strategy,
+            });
+        }
+
+        let seed_urls = raw_urls
+            .split(',')
+            .map(str::trim)
+            .filter(|url| !url.is_empty())
+            .map(ToOwned::to_owned)
+            .collect::<Vec<_>>();
+
+        if seed_urls.is_empty() {
+            return Err(RedisConfigError::MissingUrl);
+        }
+
+        let backend = match (mode, connection_type) {
+            (RedisTopology::Standalone, RedisConnectionType::Pooled) => {
+                if seed_urls.len() != 1 {
+                    return Err(RedisConfigError::MultipleStandaloneUrls);
+                }
+                RedisBackendConfig::StandalonePooled(RedisPoolSize::new(
+                    "standalone",
+                    standalone_pool_size.0,
+                    standalone_pool_size.1,
+                )?)
+            }
+            (RedisTopology::Cluster, RedisConnectionType::Pooled) => {
+                RedisBackendConfig::ClusterPooled(RedisPoolSize::new(
+                    "cluster",
+                    cluster_pool_size.0,
+                    cluster_pool_size.1,
+                )?)
+            }
+            (RedisTopology::Cluster, RedisConnectionType::Multiplexed) => {
+                RedisBackendConfig::ClusterMultiplexed
+            }
+            (mode, connection_type) => {
+                return Err(RedisConfigError::UnsupportedConnectionType {
+                    mode,
+                    connection_type,
+                });
+            }
+        };
+
+        Ok(Self {
+            mode,
+            backend,
+            seed_urls,
+            wait_timeout_ms,
+            blocking_pool_size: RedisPoolSize::new(
+                "blocking",
+                blocking_pool_size.0,
+                blocking_pool_size.1,
+            )?,
+            cache_locking_strategy,
+            read_replica_strategy,
+        })
+    }
+
+    pub fn topology(&self) -> RedisTopology {
+        self.mode
+    }
+
+    pub(crate) fn backend(&self) -> RedisBackendConfig {
+        self.backend
+    }
+
+    pub fn seed_urls(&self) -> &[String] {
+        &self.seed_urls
+    }
+
+    pub fn wait_timeout_ms(&self) -> u64 {
+        self.wait_timeout_ms
+    }
+
+    pub(crate) fn blocking_pool_size(&self) -> RedisPoolSize {
+        self.blocking_pool_size
+    }
+
+    pub fn read_replica_strategy(&self) -> ReadReplicaStrategy {
+        self.read_replica_strategy
+    }
+
+    pub fn cache_locking_strategy(&self) -> CacheLockingStrategy {
+        self.cache_locking_strategy
+    }
+}

packages/xredis/src/connection.rs

@@ -0,0 +1,321 @@
+use std::time::Duration;
+
+use futures::future::try_join_all;
+use prometheus::Registry;
+use redis::aio::ConnectionLike;
+use redis::cluster_read_routing::{
+    RandomReplicaStrategy, RoundRobinReplicaStrategy,
+};
+use redis::cluster_routing::RoutingInfo;
+use thiserror::Error;
+use tracing::warn;
+
+use crate::ReadReplicaStrategy;
+
+use super::config::{RedisBackendConfig, RedisConfig, RedisPoolSize};
+use super::metrics::{
+    LogicalPoolStatus, LogicalPoolStatusProvider, register_command_pool_metrics,
+};
+
+const POOL_RETAIN_INTERVAL: Duration = Duration::from_secs(30);
+const MAX_IDLE_CONNECTION_AGE: Duration = Duration::from_secs(5 * 60);
+const MAX_STANDALONE_CONNECTION_AGE: Duration = Duration::from_secs(120);
+
+/// The primary backing "connection provider" for a Redis backend implementation.
+#[derive(Clone)]
+pub(crate) enum RedisBackend {
+    StandalonePooled(deadpool_redis::Pool),
+    ClusterPooled(deadpool_redis::cluster::Pool),
+    ClusterMultiplexed(redis::cluster_async::ClusterConnection),
+}
+
+#[derive(Debug, Error)]
+pub enum RedisBackendBuildError {
+    #[error("failed to configure Redis client: {0}")]
+    Redis(#[from] redis::RedisError),
+    #[error("failed to build Redis pool: {0}")]
+    PoolBuild(#[from] deadpool_redis::BuildError),
+    #[error("failed to establish initial Redis pool connections: {0}")]
+    Pool(#[from] deadpool_redis::PoolError),
+}
+
+pub(crate) struct RedisConnection {
+    inner: RedisConnectionInner,
+}
+
+enum RedisConnectionInner {
+    StandalonePooled(deadpool_redis::Connection),
+    ClusterPooled(deadpool_redis::cluster::Connection),
+    ClusterMultiplexed(redis::cluster_async::ClusterConnection),
+}
+
+pub(crate) trait RoutableConnection: ConnectionLike {
+    fn route_command<'a>(
+        &'a mut self,
+        command: redis::Cmd,
+        routing: RoutingInfo,
+    ) -> redis::RedisFuture<'a, redis::Value>;
+}
+
+impl RedisBackend {
+    pub(crate) async fn new(
+        config: &RedisConfig,
+    ) -> Result<Self, RedisBackendBuildError> {
+        match config.backend() {
+            RedisBackendConfig::StandalonePooled(pool_size) => {
+                Self::standalone_pooled(config, pool_size).await
+            }
+            RedisBackendConfig::ClusterPooled(pool_size) => {
+                Self::cluster_pooled(config, pool_size).await
+            }
+            RedisBackendConfig::ClusterMultiplexed => {
+                Self::cluster_multiplexed(config).await
+            }
+        }
+    }
+
+    async fn standalone_pooled(
+        config: &RedisConfig,
+        pool_size: RedisPoolSize,
+    ) -> Result<Self, RedisBackendBuildError> {
+        let connection_config = redis::AsyncConnectionConfig::new()
+            .set_connection_timeout(None)
+            .set_response_timeout(None);
+        let manager = deadpool_redis::Manager::new_with_config(
+            config.seed_urls()[0].clone(),
+            connection_config,
+        )?;
+        let pool = deadpool_redis::Pool::builder(manager)
+            .max_size(pool_size.max())
+            .wait_timeout(Some(Duration::from_millis(config.wait_timeout_ms())))
+            .runtime(deadpool_redis::Runtime::Tokio1)
+            .build()?;
+
+        warm_standalone_pool(&pool, pool_size.min()).await?;
+        retain_standalone_pool(pool.clone());
+
+        Ok(Self::StandalonePooled(pool))
+    }
+
+    async fn cluster_pooled(
+        config: &RedisConfig,
+        pool_size: RedisPoolSize,
+    ) -> Result<Self, RedisBackendBuildError> {
+        let manager = deadpool_redis::cluster::Manager::new(
+            config.seed_urls().to_vec(),
+            false,
+        )?;
+        let pool = deadpool_redis::cluster::Pool::builder(manager)
+            .max_size(pool_size.max())
+            .wait_timeout(Some(Duration::from_millis(config.wait_timeout_ms())))
+            .runtime(deadpool_redis::Runtime::Tokio1)
+            .build()?;
+
+        if config.read_replica_strategy() != ReadReplicaStrategy::Primary {
+            warn!(
+                "Cannot respect read replica strategy when using cluster pooled backend"
+            );
+        }
+
+        warm_cluster_pool(&pool, pool_size.min()).await?;
+        retain_cluster_pool(pool.clone());
+
+        Ok(Self::ClusterPooled(pool))
+    }
+
+    async fn cluster_multiplexed(
+        config: &RedisConfig,
+    ) -> Result<Self, RedisBackendBuildError> {
+        let mut builder = redis::cluster::ClusterClientBuilder::new(
+            config.seed_urls().iter().map(String::as_str),
+        );
+
+        match config.read_replica_strategy() {
+            ReadReplicaStrategy::Primary => {}
+            ReadReplicaStrategy::RoundRobinReplica => {
+                builder = builder
+                    .read_routing_strategy(RoundRobinReplicaStrategy::new());
+            }
+            ReadReplicaStrategy::RandomReplica => {
+                builder = builder.read_routing_strategy(RandomReplicaStrategy);
+            }
+        }
+
+        let client = builder.build()?;
+        let connection = client.get_async_connection().await?;
+
+        Ok(Self::ClusterMultiplexed(connection))
+    }
+
+    pub(crate) async fn connect(
+        &self,
+    ) -> Result<RedisConnection, deadpool_redis::PoolError> {
+        let inner = match self {
+            Self::StandalonePooled(pool) => {
+                RedisConnectionInner::StandalonePooled(pool.get().await?)
+            }
+            Self::ClusterPooled(pool) => {
+                RedisConnectionInner::ClusterPooled(pool.get().await?)
+            }
+            Self::ClusterMultiplexed(connection) => {
+                RedisConnectionInner::ClusterMultiplexed(connection.clone())
+            }
+        };
+
+        Ok(RedisConnection { inner })
+    }
+
+    pub(crate) fn register_metrics(
+        &self,
+        registry: &Registry,
+    ) -> Result<(), prometheus::Error> {
+        register_command_pool_metrics(registry, self.clone())
+    }
+}
+
+impl LogicalPoolStatusProvider for RedisBackend {
+    fn logical_pool_status(&self) -> LogicalPoolStatus {
+        match self {
+            Self::StandalonePooled(pool) => {
+                LogicalPoolStatus::from_deadpool(pool.status())
+            }
+            Self::ClusterPooled(pool) => {
+                LogicalPoolStatus::from_deadpool(pool.status())
+            }
+            Self::ClusterMultiplexed(_) => {
+                LogicalPoolStatus::shared_multiplexed()
+            }
+        }
+    }
+}
+
+impl ConnectionLike for RedisConnectionInner {
+    fn req_packed_command<'a>(
+        &'a mut self,
+        cmd: &'a redis::Cmd,
+    ) -> redis::RedisFuture<'a, redis::Value> {
+        match self {
+            Self::StandalonePooled(connection) => {
+                connection.req_packed_command(cmd)
+            }
+            Self::ClusterPooled(connection) => {
+                connection.req_packed_command(cmd)
+            }
+            Self::ClusterMultiplexed(connection) => {
+                connection.req_packed_command(cmd)
+            }
+        }
+    }
+
+    fn req_packed_commands<'a>(
+        &'a mut self,
+        cmd: &'a redis::Pipeline,
+        offset: usize,
+        count: usize,
+    ) -> redis::RedisFuture<'a, Vec<redis::Value>> {
+        match self {
+            Self::StandalonePooled(connection) => {
+                connection.req_packed_commands(cmd, offset, count)
+            }
+            Self::ClusterPooled(connection) => {
+                connection.req_packed_commands(cmd, offset, count)
+            }
+            Self::ClusterMultiplexed(connection) => {
+                connection.req_packed_commands(cmd, offset, count)
+            }
+        }
+    }
+
+    fn get_db(&self) -> i64 {
+        match self {
+            Self::StandalonePooled(connection) => connection.get_db(),
+            Self::ClusterPooled(connection) => connection.get_db(),
+            Self::ClusterMultiplexed(connection) => connection.get_db(),
+        }
+    }
+}
+
+impl ConnectionLike for RedisConnection {
+    fn req_packed_command<'a>(
+        &'a mut self,
+        cmd: &'a redis::Cmd,
+    ) -> redis::RedisFuture<'a, redis::Value> {
+        self.inner.req_packed_command(cmd)
+    }
+
+    fn req_packed_commands<'a>(
+        &'a mut self,
+        cmd: &'a redis::Pipeline,
+        offset: usize,
+        count: usize,
+    ) -> redis::RedisFuture<'a, Vec<redis::Value>> {
+        self.inner.req_packed_commands(cmd, offset, count)
+    }
+
+    fn get_db(&self) -> i64 {
+        self.inner.get_db()
+    }
+}
+
+impl RoutableConnection for RedisConnection {
+    fn route_command<'a>(
+        &'a mut self,
+        command: redis::Cmd,
+        routing: RoutingInfo,
+    ) -> redis::RedisFuture<'a, redis::Value> {
+        Box::pin(async move {
+            match &mut self.inner {
+                RedisConnectionInner::StandalonePooled(connection) => {
+                    command.query_async(connection).await
+                }
+                RedisConnectionInner::ClusterPooled(connection) => {
+                    command.query_async(connection).await
+                }
+                RedisConnectionInner::ClusterMultiplexed(connection) => {
+                    connection.route_command(command, routing).await
+                }
+            }
+        })
+    }
+}
+
+async fn warm_standalone_pool(
+    pool: &deadpool_redis::Pool,
+    min: usize,
+) -> Result<(), deadpool_redis::PoolError> {
+    let connections = try_join_all((0..min).map(|_| pool.get())).await?;
+    drop(connections);
+    Ok(())
+}
+
+async fn warm_cluster_pool(
+    pool: &deadpool_redis::cluster::Pool,
+    min: usize,
+) -> Result<(), deadpool_redis::PoolError> {
+    let connections = try_join_all((0..min).map(|_| pool.get())).await?;
+    drop(connections);
+    Ok(())
+}
+
+fn retain_standalone_pool(pool: deadpool_redis::Pool) {
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(POOL_RETAIN_INTERVAL).await;
+            pool.retain(|_, metrics| {
+                metrics.last_used() < MAX_IDLE_CONNECTION_AGE
+                    && metrics.created.elapsed() < MAX_STANDALONE_CONNECTION_AGE
+            });
+        }
+    });
+}
+
+fn retain_cluster_pool(pool: deadpool_redis::cluster::Pool) {
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(POOL_RETAIN_INTERVAL).await;
+            pool.retain(|_, metrics| {
+                metrics.last_used() < MAX_IDLE_CONNECTION_AGE
+            });
+        }
+    });
+}

packages/xredis/src/key.rs

@@ -0,0 +1,75 @@
+use std::fmt::Display;
+use std::sync::Arc;
+
+use crate::RedisTopology;
+
+#[derive(Debug, Clone)]
+pub struct KeyBuilder {
+    meta_namespace: Arc<str>,
+    mode: RedisTopology,
+}
+
+impl KeyBuilder {
+    pub fn new(
+        meta_namespace: impl Into<Arc<str>>,
+        mode: RedisTopology,
+    ) -> Self {
+        Self {
+            meta_namespace: meta_namespace.into(),
+            mode,
+        }
+    }
+
+    /// Build a key with the given namespace and logical key. The logical key is used as the key's slot tag.
+    pub fn entity(&self, namespace: &str, logical_key: impl Display) -> String {
+        let logical_key = logical_key.to_string();
+        self.with_slot(namespace, &logical_key, &logical_key)
+    }
+
+    /// Build a metadata key with the given namespace and logical key. The slot tag is fixed to `_metadata`.
+    pub fn metadata(
+        &self,
+        namespace: &str,
+        logical_key: impl Display,
+    ) -> String {
+        self.with_slot(namespace, logical_key, "_metadata")
+    }
+
+    /// Build a key with the given namespace, logical key, and slot tag.
+    pub fn with_slot(
+        &self,
+        namespace: &str,
+        logical_key: impl Display,
+        slot_tag: impl Display,
+    ) -> String {
+        match self.mode {
+            RedisTopology::Standalone => {
+                format!("{}_{}:{}", self.meta_namespace, namespace, logical_key)
+            }
+            RedisTopology::Cluster => format!(
+                "{}_{}:{{{}}}:{}",
+                self.meta_namespace,
+                namespace,
+                escape_slot_tag(&slot_tag.to_string()),
+                logical_key
+            ),
+        }
+    }
+}
+
+fn escape_slot_tag(value: &str) -> String {
+    if value.is_empty() {
+        return "%00".to_string();
+    }
+
+    let mut escaped = String::with_capacity(value.len());
+    for character in value.chars() {
+        match character {
+            '%' => escaped.push_str("%25"),
+            '{' => escaped.push_str("%7B"),
+            '}' => escaped.push_str("%7D"),
+            _ => escaped.push(character),
+        }
+    }
+    escaped
+}

packages/xredis/src/lib.rs

@@ -0,0 +1,386 @@
+#![recursion_limit = "256"]
+
+use std::fmt::{Debug, Display};
+use std::future::Future;
+use std::hash::Hash;
+use std::sync::Arc;
+
+use dashmap::DashMap;
+use prometheus::Registry;
+use redis::aio::ConnectionLike;
+use redis::{FromRedisValue, ToRedisArgs};
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+
+mod blocking;
+mod cache;
+mod commands;
+mod config;
+mod connection;
+mod key;
+mod metrics;
+mod pubsub;
+mod routing;
+mod util;
+
+use cache::{CacheManager, ConnectionProvider};
+pub use cache::{
+    CacheSettings, Codec, EncodingFormat, InvalidCodec, InvalidEncodingFormat,
+    RedisValue,
+};
+pub use config::{
+    CacheLockingStrategy, InvalidCacheLockingStrategy,
+    InvalidRedisConnectionType, InvalidRedisMode,
+    InvalidRedisReadReplicaStrategy, ReadReplicaStrategy, RedisConfig,
+    RedisConfigError, RedisConnectionType, RedisTopology,
+};
+use connection::RedisBackend;
+pub use connection::RedisBackendBuildError;
+pub use key::KeyBuilder;
+use thiserror::Error as ThisError;
+
+#[derive(Debug, ThisError)]
+pub enum Error {
+    #[error("error while interacting with Redis: {0}")]
+    Redis(#[from] redis::RedisError),
+    #[error("Redis pool error: {0}")]
+    Pool(#[from] deadpool_redis::PoolError),
+    #[error("error while serializing a Redis cache value: {0}")]
+    Serialization(#[from] serde_json::Error),
+    #[error("Redis blocking timeout must be greater than zero")]
+    InvalidBlockingTimeout,
+    #[error(
+        "timeout waiting on local cache lock ({released}/{total} released)"
+    )]
+    LocalCacheTimeout { released: usize, total: usize },
+}
+
+#[derive(Clone)]
+pub struct RedisPool {
+    backend: RedisBackend,
+    blocking: blocking::RedisBlockingPool,
+    cache: CacheManager,
+    config: RedisConfig,
+    key_builder: KeyBuilder,
+}
+
+pub struct RedisConnection {
+    inner: connection::RedisConnection,
+    key_builder: KeyBuilder,
+    settings: CacheSettings,
+}
+
+impl RedisPool {
+    pub async fn new(
+        meta_namespace: impl Into<Arc<str>>,
+        config: RedisConfig,
+        cache_settings: CacheSettings,
+    ) -> Result<Self, RedisBackendBuildError> {
+        tracing::info!(
+            strategy = %config.cache_locking_strategy(),
+            "configured Redis cache locking"
+        );
+
+        let backend = RedisBackend::new(&config).await?;
+
+        let blocking = blocking::RedisBlockingPool::new(&config).await?;
+        let key_builder = KeyBuilder::new(meta_namespace, config.topology());
+        let cache = CacheManager::new(key_builder.clone(), cache_settings);
+
+        Ok(Self {
+            backend,
+            blocking,
+            cache,
+            config,
+            key_builder,
+        })
+    }
+
+    pub fn key(&self) -> &KeyBuilder {
+        &self.key_builder
+    }
+}
+
+impl RedisPool {
+    pub async fn connect(&self) -> Result<RedisConnection, Error> {
+        Ok(RedisConnection {
+            inner: self.backend.connect().await?,
+            key_builder: self.key_builder.clone(),
+            settings: self.cache.settings().clone(),
+        })
+    }
+
+    pub async fn register_and_set_metrics(
+        &self,
+        registry: &Registry,
+    ) -> Result<(), prometheus::Error> {
+        self.backend.register_metrics(registry)?;
+        self.blocking.register_metrics(registry)
+    }
+
+    pub async fn get_cached_keys<F, Fut, T, K, E>(
+        &self,
+        namespace: &str,
+        keys: &[K],
+        closure: F,
+    ) -> Result<Vec<T>, E>
+    where
+        F: FnOnce(Vec<K>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, T>, E>>,
+        E: From<Error>,
+        T: Serialize + DeserializeOwned,
+        K: Display
+            + Hash
+            + Eq
+            + PartialEq
+            + Clone
+            + DeserializeOwned
+            + Serialize
+            + Debug,
+    {
+        self.cache
+            .get_cached_keys(self, namespace, keys, closure)
+            .await
+    }
+
+    pub async fn get_cached_keys_raw<F, Fut, T, K, E>(
+        &self,
+        namespace: &str,
+        keys: &[K],
+        closure: F,
+    ) -> Result<std::collections::HashMap<K, T>, E>
+    where
+        F: FnOnce(Vec<K>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, T>, E>>,
+        E: From<Error>,
+        T: Serialize + DeserializeOwned,
+        K: Display
+            + Hash
+            + Eq
+            + PartialEq
+            + Clone
+            + DeserializeOwned
+            + Serialize
+            + Debug,
+    {
+        self.cache
+            .get_cached_keys_raw(self, namespace, keys, closure)
+            .await
+    }
+
+    pub async fn get_cached_keys_with_slug<F, Fut, T, I, K, S, E>(
+        &self,
+        namespace: &str,
+        slug_namespace: &str,
+        case_sensitive: bool,
+        keys: &[I],
+        closure: F,
+    ) -> Result<Vec<T>, E>
+    where
+        F: FnOnce(Vec<I>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, (Option<S>, T)>, E>>,
+        E: From<Error>,
+        T: Serialize + DeserializeOwned,
+        I: Display + Hash + Eq + PartialEq + Clone + Debug,
+        K: Display
+            + Hash
+            + Eq
+            + PartialEq
+            + Clone
+            + DeserializeOwned
+            + Serialize,
+        S: Display + Clone + DeserializeOwned + Serialize + Debug,
+    {
+        self.cache
+            .get_cached_keys_with_slug(
+                self,
+                namespace,
+                slug_namespace,
+                case_sensitive,
+                keys,
+                closure,
+            )
+            .await
+    }
+
+    pub async fn get_cached_keys_raw_with_slug<F, Fut, T, I, K, S, E>(
+        &self,
+        namespace: &str,
+        slug_namespace: Option<&str>,
+        case_sensitive: bool,
+        keys: &[I],
+        closure: F,
+    ) -> Result<std::collections::HashMap<K, T>, E>
+    where
+        F: FnOnce(Vec<I>) -> Fut,
+        Fut: Future<Output = Result<DashMap<K, (Option<S>, T)>, E>>,
+        E: From<Error>,
+        T: Serialize + DeserializeOwned,
+        I: Display + Hash + Eq + PartialEq + Clone + Debug,
+        K: Display
+            + Hash
+            + Eq
+            + PartialEq
+            + Clone
+            + DeserializeOwned
+            + Serialize,
+        S: Display + Clone + DeserializeOwned + Serialize + Debug,
+    {
+        self.cache
+            .get_cached_keys_raw_with_slug(
+                self,
+                namespace,
+                slug_namespace,
+                case_sensitive,
+                keys,
+                closure,
+            )
+            .await
+    }
+}
+
+impl ConnectionProvider for RedisPool {
+    type Connection = RedisConnection;
+
+    fn connect(
+        &self,
+    ) -> impl Future<Output = Result<Self::Connection, Error>> + Send {
+        RedisPool::connect(self)
+    }
+}
+
+impl RedisConnection {
+    pub fn key(&self) -> &KeyBuilder {
+        &self.key_builder
+    }
+
+    pub async fn set<D>(
+        &mut self,
+        key: &str,
+        data: D,
+        expiry: Option<i64>,
+    ) -> Result<(), Error>
+    where
+        D: ToRedisArgs + Send + Sync + Debug,
+    {
+        commands::set(
+            &mut self.inner,
+            key,
+            data,
+            expiry.unwrap_or(self.settings.default_expiry),
+        )
+        .await
+    }
+
+    pub async fn set_serialized<D>(
+        &mut self,
+        key: &str,
+        data: D,
+        expiry: Option<i64>,
+    ) -> Result<(), Error>
+    where
+        D: Serialize,
+    {
+        commands::set_serialized(
+            &mut self.inner,
+            key,
+            data,
+            expiry,
+            &self.settings,
+        )
+        .await
+    }
+
+    pub async fn get(&mut self, key: &str) -> Result<Option<String>, Error> {
+        commands::get(&mut self.inner, key).await
+    }
+
+    pub async fn get_many(
+        &mut self,
+        keys: &[String],
+    ) -> Result<Vec<Option<Vec<u8>>>, Error> {
+        commands::get_many(&mut self.inner, keys).await
+    }
+
+    pub async fn get_many_typed<R>(
+        &mut self,
+        keys: &[String],
+    ) -> Result<Vec<Option<R>>, Error>
+    where
+        R: FromRedisValue,
+    {
+        commands::get_many_as(&mut self.inner, keys).await
+    }
+
+    pub async fn get_deserialized<R>(
+        &mut self,
+        key: &str,
+    ) -> Result<Option<R>, Error>
+    where
+        R: for<'a> serde::Deserialize<'a>,
+    {
+        commands::get_deserialized(&mut self.inner, key, &self.settings).await
+    }
+
+    pub async fn get_many_deserialized<R>(
+        &mut self,
+        keys: &[String],
+    ) -> Result<Vec<Option<R>>, Error>
+    where
+        R: for<'a> serde::Deserialize<'a>,
+    {
+        commands::get_many_deserialized(&mut self.inner, keys, &self.settings)
+            .await
+    }
+
+    pub async fn delete(&mut self, key: &str) -> Result<(), Error> {
+        commands::delete(&mut self.inner, key).await
+    }
+
+    pub async fn delete_many(&mut self, keys: &[String]) -> Result<(), Error> {
+        commands::delete_many(&mut self.inner, keys).await
+    }
+
+    pub async fn lpush<D>(&mut self, key: &str, value: D) -> Result<(), Error>
+    where
+        D: ToRedisArgs + Send + Sync + Debug,
+    {
+        commands::lpush(&mut self.inner, key, value).await
+    }
+
+    pub async fn incr(&mut self, key: &str) -> Result<Option<u64>, Error> {
+        commands::incr(&mut self.inner, key).await
+    }
+}
+
+impl ConnectionLike for RedisConnection {
+    fn req_packed_command<'a>(
+        &'a mut self,
+        command: &'a redis::Cmd,
+    ) -> redis::RedisFuture<'a, redis::Value> {
+        self.inner.req_packed_command(command)
+    }
+
+    fn req_packed_commands<'a>(
+        &'a mut self,
+        pipeline: &'a redis::Pipeline,
+        offset: usize,
+        count: usize,
+    ) -> redis::RedisFuture<'a, Vec<redis::Value>> {
+        self.inner.req_packed_commands(pipeline, offset, count)
+    }
+
+    fn get_db(&self) -> i64 {
+        self.inner.get_db()
+    }
+}
+
+impl connection::RoutableConnection for RedisConnection {
+    fn route_command<'a>(
+        &'a mut self,
+        command: redis::Cmd,
+        routing: redis::cluster_routing::RoutingInfo,
+    ) -> redis::RedisFuture<'a, redis::Value> {
+        self.inner.route_command(command, routing)
+    }
+}

packages/xredis/src/metrics.rs

@@ -0,0 +1,161 @@
+use std::time::Duration;
+
+use prometheus::{IntGauge, Registry};
+
+const METRICS_UPDATE_INTERVAL: Duration = Duration::from_secs(5);
+
+#[derive(Debug, Clone, Copy)]
+pub(super) struct LogicalPoolStatus {
+    max_size: usize,
+    size: usize,
+    available: usize,
+    waiting: usize,
+}
+
+impl LogicalPoolStatus {
+    pub(super) fn from_deadpool(status: deadpool_redis::Status) -> Self {
+        Self {
+            max_size: status.max_size,
+            size: status.size,
+            available: status.available,
+            waiting: status.waiting,
+        }
+    }
+
+    pub(super) fn shared_multiplexed() -> Self {
+        Self {
+            max_size: 1,
+            size: 1,
+            available: 1,
+            waiting: 0,
+        }
+    }
+}
+
+pub(super) trait LogicalPoolStatusProvider:
+    Clone + Send + 'static
+{
+    fn logical_pool_status(&self) -> LogicalPoolStatus;
+}
+
+#[derive(Debug, Clone, Copy)]
+enum RedisPoolMetricsKind {
+    Command,
+    Blocking,
+}
+
+impl RedisPoolMetricsKind {
+    fn metric_prefix(self) -> &'static str {
+        match self {
+            Self::Command => "labrinth_redis_pool",
+            Self::Blocking => "labrinth_redis_blocking_pool",
+        }
+    }
+
+    fn description(self) -> &'static str {
+        match self {
+            Self::Command => "Redis command pool",
+            Self::Blocking => "Redis blocking-command pool",
+        }
+    }
+}
+
+#[derive(Clone)]
+struct RedisPoolMetrics {
+    max_size: IntGauge,
+    size: IntGauge,
+    available: IntGauge,
+    waiting: IntGauge,
+}
+
+impl RedisPoolMetrics {
+    fn register(
+        registry: &Registry,
+        kind: RedisPoolMetricsKind,
+    ) -> Result<Self, prometheus::Error> {
+        let prefix = kind.metric_prefix();
+        let description = kind.description();
+        let max_size = IntGauge::new(
+            format!("{prefix}_max_size"),
+            format!(
+                "Maximum logical connection count for the {description}; clustered logical connections may own multiple physical sockets"
+            ),
+        )?;
+        let size = IntGauge::new(
+            format!("{prefix}_size"),
+            format!(
+                "Current logical connection count for the {description}; clustered logical connections may own multiple physical sockets"
+            ),
+        )?;
+        let available = IntGauge::new(
+            format!("{prefix}_available"),
+            format!("Available logical connections in the {description}"),
+        )?;
+        let waiting = IntGauge::new(
+            format!("{prefix}_waiting"),
+            format!(
+                "Number of futures waiting for a logical connection from the {description}"
+            ),
+        )?;
+
+        registry.register(Box::new(max_size.clone()))?;
+        registry.register(Box::new(size.clone()))?;
+        registry.register(Box::new(available.clone()))?;
+        registry.register(Box::new(waiting.clone()))?;
+
+        Ok(Self {
+            max_size,
+            size,
+            available,
+            waiting,
+        })
+    }
+
+    fn set(&self, status: LogicalPoolStatus) {
+        self.max_size.set(status.max_size as i64);
+        self.size.set(status.size as i64);
+        self.available.set(status.available as i64);
+        self.waiting.set(status.waiting as i64);
+    }
+}
+
+pub(super) fn register_command_pool_metrics<P>(
+    registry: &Registry,
+    provider: P,
+) -> Result<(), prometheus::Error>
+where
+    P: LogicalPoolStatusProvider,
+{
+    register_pool_metrics(registry, RedisPoolMetricsKind::Command, provider)
+}
+
+pub(super) fn register_blocking_pool_metrics<P>(
+    registry: &Registry,
+    provider: P,
+) -> Result<(), prometheus::Error>
+where
+    P: LogicalPoolStatusProvider,
+{
+    register_pool_metrics(registry, RedisPoolMetricsKind::Blocking, provider)
+}
+
+fn register_pool_metrics<P>(
+    registry: &Registry,
+    kind: RedisPoolMetricsKind,
+    provider: P,
+) -> Result<(), prometheus::Error>
+where
+    P: LogicalPoolStatusProvider,
+{
+    let metrics = RedisPoolMetrics::register(registry, kind)?;
+    metrics.set(provider.logical_pool_status());
+
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(METRICS_UPDATE_INTERVAL).await;
+            metrics.set(provider.logical_pool_status());
+        }
+    });
+
+    Ok(())
+}

packages/xredis/src/pubsub.rs

@@ -0,0 +1,136 @@
+use std::time::Duration;
+
+use futures::StreamExt;
+use redis::ToRedisArgs;
+use tokio::sync::mpsc;
+use tracing::{info, warn};
+
+use super::{Error, RedisPool};
+
+const PUBSUB_BUFFER_SIZE: usize = 1024;
+const INITIAL_RECONNECT_BACKOFF: Duration = Duration::from_millis(250);
+const MAX_RECONNECT_BACKOFF: Duration = Duration::from_secs(30);
+
+enum SubscriptionOutcome {
+    SubscriberClosed,
+    Disconnected,
+}
+
+impl RedisPool {
+    pub fn subscribe(&self, channel: &'static str) -> mpsc::Receiver<Vec<u8>> {
+        let seed_urls = self.config.seed_urls().to_vec();
+        let (sender, receiver) = mpsc::channel(PUBSUB_BUFFER_SIZE);
+        tokio::spawn(run_subscription(seed_urls, channel, sender));
+        receiver
+    }
+
+    #[doc(hidden)]
+    #[cfg(feature = "test")]
+    pub fn subscribe_with_seed_urls(
+        seed_urls: Vec<String>,
+        channel: &'static str,
+    ) -> mpsc::Receiver<Vec<u8>> {
+        let (sender, receiver) = mpsc::channel(PUBSUB_BUFFER_SIZE);
+        tokio::spawn(run_subscription(seed_urls, channel, sender));
+        receiver
+    }
+
+    pub async fn publish<M>(
+        &self,
+        channel: &str,
+        message: M,
+    ) -> Result<(), Error>
+    where
+        M: ToRedisArgs + Send + Sync,
+    {
+        let mut connection = self.connect().await?;
+        let _: usize = redis::cmd("PUBLISH")
+            .arg(channel)
+            .arg(message)
+            .query_async(&mut connection)
+            .await
+            .map_err(Error::from)?;
+        Ok(())
+    }
+}
+
+async fn run_subscription(
+    seed_urls: Vec<String>,
+    channel: &'static str,
+    sender: mpsc::Sender<Vec<u8>>,
+) {
+    let mut next_seed = 0;
+    let mut reconnect_backoff = INITIAL_RECONNECT_BACKOFF;
+
+    loop {
+        let mut connected = false;
+
+        for _ in 0..seed_urls.len() {
+            let seed_url = &seed_urls[next_seed];
+            next_seed = (next_seed + 1) % seed_urls.len();
+
+            match forward_from_seed(seed_url, channel, &sender).await {
+                Ok(SubscriptionOutcome::SubscriberClosed) => return,
+                Ok(SubscriptionOutcome::Disconnected) => {
+                    warn!(channel, "Redis Pub/Sub connection disconnected");
+                    connected = true;
+                    break;
+                }
+                Err(error) => {
+                    warn!(
+                        ?error,
+                        channel,
+                        "Failed to establish Redis Pub/Sub subscription"
+                    );
+                }
+            }
+        }
+
+        if sender.is_closed() {
+            return;
+        }
+
+        let delay = if connected {
+            INITIAL_RECONNECT_BACKOFF
+        } else {
+            reconnect_backoff
+        };
+        tokio::time::sleep(delay).await;
+
+        reconnect_backoff = if connected {
+            INITIAL_RECONNECT_BACKOFF
+        } else {
+            reconnect_backoff
+                .saturating_mul(2)
+                .min(MAX_RECONNECT_BACKOFF)
+        };
+    }
+}
+
+async fn forward_from_seed(
+    seed_url: &str,
+    channel: &'static str,
+    sender: &mpsc::Sender<Vec<u8>>,
+) -> redis::RedisResult<SubscriptionOutcome> {
+    let client = redis::Client::open(seed_url)?;
+    let mut pubsub = client.get_async_pubsub().await?;
+    pubsub.subscribe(channel).await?;
+    info!(channel, "Established Redis Pub/Sub subscription");
+
+    let mut stream = pubsub.into_on_message();
+    while let Some(message) = stream.next().await {
+        if message.get_channel_name() != channel {
+            continue;
+        }
+
+        if sender
+            .send(message.get_payload_bytes().to_vec())
+            .await
+            .is_err()
+        {
+            return Ok(SubscriptionOutcome::SubscriberClosed);
+        }
+    }
+
+    Ok(SubscriptionOutcome::Disconnected)
+}

packages/xredis/src/routing.rs

@@ -0,0 +1,29 @@
+use std::collections::HashMap;
+
+use redis::cluster_routing::{
+    MultiSlotArgPattern, MultipleNodeRoutingInfo, ResponsePolicy, Route,
+    RoutingInfo, SlotAddr,
+};
+
+/// Returns a routing specification to split an MGET by hash slots and
+/// route to **primaries only**.
+///
+/// Use this if you need a command to be routed to primaries only.
+///
+/// It's not needed just to split an MGET; redis-rs already does that by default.
+pub(crate) fn primary_mget_routing<K: AsRef<[u8]>>(keys: &[K]) -> RoutingInfo {
+    let mut keys_by_slot: HashMap<Route, Vec<usize>> = HashMap::new();
+
+    for (index, key) in keys.iter().enumerate() {
+        let route = Route::with_key(key.as_ref(), SlotAddr::Master);
+        keys_by_slot.entry(route).or_default().push(index);
+    }
+
+    RoutingInfo::MultiNode((
+        MultipleNodeRoutingInfo::MultiSlot((
+            keys_by_slot.into_iter().collect(),
+            MultiSlotArgPattern::KeysOnly,
+        )),
+        Some(ResponsePolicy::CombineArrays),
+    ))
+}

packages/xredis/src/util.rs

@@ -0,0 +1,37 @@
+use redis::{FromRedisValue, RedisResult, ToRedisArgs};
+use tracing::{Instrument, info_span};
+
+pub(crate) fn cmd(name: &str) -> InstrumentedCmd {
+    InstrumentedCmd {
+        inner: redis::cmd(name),
+        name: name.to_string(),
+    }
+}
+
+pub(crate) struct InstrumentedCmd {
+    inner: redis::Cmd,
+    name: String,
+}
+
+impl InstrumentedCmd {
+    #[inline]
+    pub fn arg<T: ToRedisArgs>(&mut self, arg: T) -> &mut Self {
+        self.inner.arg(arg);
+        self
+    }
+
+    #[inline]
+    pub async fn query_async<T: FromRedisValue>(
+        &self,
+        con: &mut impl redis::aio::ConnectionLike,
+    ) -> RedisResult<T> {
+        let span = info_span!(
+            "cmd.query_async",
+            // <https://opentelemetry.io/docs/specs/semconv/db/redis/>
+            db.system.name = "redis",
+            db.operation.name = self.name,
+            db.query.text = self.name,
+        );
+        self.inner.query_async(con).instrument(span).await
+    }
+}