redo error handling in xredis

Cargo.lock

@@ -13443,12 +13443,13 @@ dependencies = [
  "chrono",
  "dashmap",
  "deadpool-redis",
+ "eyre",
  "futures",
  "lz4_flex",
+ "postcard",
  "prometheus",
  "redis",
  "serde",
- "serde_json",
  "thiserror 2.0.17",
  "tokio",
  "tracing",

apps/labrinth/AGENTS.md

@@ -12,8 +12,10 @@
   - no trailing punctuation
   - wrap code items e.g. type names in backticks
 - Prefer `wrap_internal_err`, `wrap_request_err` when attaching context to an existing error (like Anyhow `context` or Eyre `wrap_err`)
+- Prefer importing `eyre::Result` and using `Result<T>` instead of `eyre::Result<T>`
+- Prefer `eyre::Ok(value)` instead of `Ok::<_, eyre::Report>(value)` when an explicit Eyre result type is needed
 - All operations should ideally have some context attached
-  - Database operations can have a message like `.wrap_internal_err("failed to fetch XYZ")`
+  - Database operations can have a message like `.wrap_internal_err("fetching XYZ")`
 - You can perform real-time queries against the databases in the Docker Compose
   - `docker exec labrinth-postgres psql -c "select 1"`
   - `docker exec labrinth-redis redis-cli flushall`

apps/labrinth/src/auth/mod.rs

@@ -59,12 +59,6 @@ pub enum AuthenticationError {
     Url,
 }
 
-impl From<xredis::Error> for AuthenticationError {
-    fn from(error: xredis::Error) -> Self {
-        Self::Database(error.into())
-    }
-}
-
 impl actix_web::ResponseError for AuthenticationError {
     fn status_code(&self) -> StatusCode {
         match self {

apps/labrinth/src/database/models/mod.rs

@@ -77,8 +77,6 @@ pub enum DatabaseError {
     SerdeCacheError(#[from] serde_json::Error),
     #[error("error while encoding or decoding the cache: {0}")]
     PostcardCacheError(#[from] postcard::Error),
-    #[error(transparent)]
-    Redis(#[from] xredis::Error),
     #[error("Schema error: {0}")]
     SchemaError(String),
 }

apps/labrinth/src/database/models/notifications_template_item.rs

@@ -1,6 +1,7 @@
 use crate::database::models::DatabaseError;
 use crate::models::v3::notifications::{NotificationChannel, NotificationType};
 use crate::routes::ApiError;
+use crate::util::error::Context;
 use serde::{Deserialize, Serialize};
 use xredis::RedisPool;
 
@@ -123,12 +124,16 @@ where
         html: String,
     }
 
-    let mut redis_conn = redis.connect().await?;
+    let mut redis_conn = redis.connect().await.wrap_internal_err(
+        "connecting to redis for dynamic notification html",
+    )?;
     let redis_key = redis_conn
         .key()
         .metadata(TEMPLATES_DYNAMIC_HTML_NAMESPACE, key);
-    if let Some(body) =
+    if let Some(body) = redis_conn
-        redis_conn.get_deserialized::<HtmlBody>(&redis_key).await?
+        .get_deserialized::<HtmlBody>(&redis_key)
+        .await
+        .wrap_internal_err("fetching dynamic notification html from redis")?
     {
         return Ok(body.html);
     }
@@ -136,14 +141,17 @@ where
     drop(redis_conn);
 
     let cached = HtmlBody { html: get().await? };
-    let mut redis_conn = redis.connect().await?;
+    let mut redis_conn = redis.connect().await.wrap_internal_err(
+        "connecting to redis for dynamic notification html",
+    )?;
     let redis_key = redis_conn
         .key()
         .metadata(TEMPLATES_DYNAMIC_HTML_NAMESPACE, key);
 
     redis_conn
         .set_serialized(&redis_key, &cached, Some(HTML_DATA_CACHE_EXPIRY))
-        .await?;
+        .await
+        .wrap_internal_err("writing dynamic notification html to redis")?;
 
     Ok(cached.html)
 }

apps/labrinth/src/database/models/project_item.rs

@@ -946,7 +946,7 @@ impl DBProject {
             },
         )
         .await
-        .wrap_internal_err("failed to fetch cached projects")?;
+        .wrap_internal_err("fetching cached projects")?;
 
         Ok(val)
     }

apps/labrinth/src/database/redis.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use crate::env::ENV;
+use eyre::{Result, WrapErr};
 
 struct RedisConfig {
     inner: xredis::RedisConfig,
@@ -8,7 +9,7 @@ struct RedisConfig {
 }
 
 impl RedisConfig {
-    fn from_env() -> Result<Self, xredis::RedisConfigError> {
+    fn from_env() -> Result<Self> {
         let inner = xredis::RedisConfig::new(
             ENV.REDIS_TOPOLOGY,
             ENV.REDIS_CONNECTION_TYPE,
@@ -25,7 +26,8 @@ impl RedisConfig {
             (ENV.REDIS_BLOCKING_MAX_CONNECTIONS as usize, 0),
             ENV.REDIS_CACHE_LOCKING_STRATEGY,
             ENV.REDIS_READ_REPLICA_STRATEGY,
-        )?;
+        )
+        .wrap_err("loading Redis configuration from environment")?;
         let cache_settings = xredis::CacheSettings {
             default_expiry: ENV.REDIS_DEFAULT_EXPIRY,
             actual_expiry: ENV.REDIS_ACTUAL_EXPIRY,

apps/labrinth/src/env.rs

@@ -165,7 +165,7 @@ vars! {
     REDIS_BLOCKING_MAX_CONNECTIONS: u32 = 256u32;
 
     // The encoding format used for Redis cache values.
-    REDIS_ENCODING_FORMAT: xredis::EncodingFormat = xredis::EncodingFormat::Json;
+    REDIS_ENCODING_FORMAT: xredis::EncodingFormat = xredis::EncodingFormat::Postcard;
     // The level of LZ4 compression used for Redis cache values. A value of 0 disables compression (supports 1-12)
     REDIS_COMPRESSION_LEVEL: i32 = 0i32;
     // The compression algorithm used for Redis cache values. Currently only LZ4 is supported.

apps/labrinth/src/queue/analytics/mod.rs

@@ -4,6 +4,7 @@ use crate::models::analytics::{
 };
 use crate::routes::ApiError;
 use crate::routes::analytics::MINECRAFT_SERVER_PLAYS;
+use crate::util::error::Context;
 use dashmap::{DashMap, DashSet};
 use std::collections::HashMap;
 use tracing::trace;
@@ -142,10 +143,15 @@ impl AnalyticsQueue {
                     )
                 })
                 .collect::<Vec<_>>();
-            let mut redis_connection = redis.connect().await?;
+            let mut redis_connection =
+                redis.connect().await.wrap_internal_err(
+                    "connecting to redis for server play counts",
+                )?;
 
-            let results =
+            let results = redis_connection
-                redis_connection.get_many_typed::<u32>(&redis_keys).await?;
+                .get_many_typed::<u32>(&redis_keys)
+                .await
+                .wrap_internal_err("fetching server play counts from redis")?;
             for (idx, count) in results.into_iter().enumerate() {
                 let new_count = if let Some(count) = count {
                     if count >= MINECRAFT_SERVER_PLAYS_LIMIT {
@@ -164,7 +170,8 @@ impl AnalyticsQueue {
                         new_count,
                         Some(MINECRAFT_SERVER_PLAYS_EXPIRY as i64),
                     )
-                    .await?;
+                    .await
+                    .wrap_internal_err("writing server play count to redis")?;
             }
 
             let mut plays = client
@@ -198,10 +205,15 @@ impl AnalyticsQueue {
                     )
                 })
                 .collect::<Vec<_>>();
-            let mut redis_connection = redis.connect().await?;
+            let mut redis_connection = redis
+                .connect()
+                .await
+                .wrap_internal_err("connecting to redis for view counts")?;
 
-            let results =
+            let results = redis_connection
-                redis_connection.get_many_typed::<u32>(&redis_keys).await?;
+                .get_many_typed::<u32>(&redis_keys)
+                .await
+                .wrap_internal_err("fetching view counts from redis")?;
             for (idx, count) in results.into_iter().enumerate() {
                 let new_count =
                     if let Some((views, monetized)) = raw_views.get_mut(idx) {
@@ -226,7 +238,8 @@ impl AnalyticsQueue {
                 let key = &redis_keys[idx];
                 redis_connection
                     .set(key, new_count, Some(6 * 60 * 60))
-                    .await?;
+                    .await
+                    .wrap_internal_err("writing view count to redis")?;
             }
 
             let mut views = client.insert::<PageView>("views").await?;
@@ -267,10 +280,15 @@ impl AnalyticsQueue {
                     )
                 })
                 .collect::<Vec<_>>();
-            let mut redis_connection = redis.connect().await?;
+            let mut redis_connection = redis
+                .connect()
+                .await
+                .wrap_internal_err("connecting to redis for download counts")?;
 
-            let results =
+            let results = redis_connection
-                redis_connection.get_many_typed::<u32>(&redis_keys).await?;
+                .get_many_typed::<u32>(&redis_keys)
+                .await
+                .wrap_internal_err("fetching download counts from redis")?;
             for (idx, count) in results.into_iter().enumerate() {
                 let new_count = if let Some(count) = count {
                     if count > 5 {
@@ -286,7 +304,8 @@ impl AnalyticsQueue {
                 let key = &redis_keys[idx];
                 redis_connection
                     .set(key, new_count, Some(6 * 60 * 60))
-                    .await?;
+                    .await
+                    .wrap_internal_err("writing download count to redis")?;
             }
 
             let mut transaction = pool.begin().await?;

apps/labrinth/src/routes/mod.rs

@@ -267,12 +267,6 @@ pub enum ApiError {
     },
 }
 
-impl From<xredis::Error> for ApiError {
-    fn from(error: xredis::Error) -> Self {
-        Self::Database(error.into())
-    }
-}
-
 impl ApiError {
     pub fn delphi(err: impl Into<eyre::Error>) -> Self {
         Self::Delphi(err.into())

packages/xredis/Cargo.toml

@@ -10,6 +10,7 @@ ariadne = { workspace = true }
 chrono = { workspace = true }
 dashmap = { workspace = true }
 deadpool-redis = { workspace = true, features = ["cluster-async"] }
+eyre = { workspace = true }
 futures = { workspace = true }
 lz4_flex = { workspace = true }
 prometheus = { workspace = true }
@@ -20,7 +21,7 @@ redis = { workspace = true, features = [
   "tokio-comp"
 ] }
 serde = { workspace = true, features = ["derive"] }
-serde_json = { workspace = true }
+postcard = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["rt", "sync", "time"] }
 tracing = { workspace = true }

packages/xredis/src/blocking.rs

@@ -1,14 +1,14 @@
 use std::time::Duration;
 
+use eyre::{Result, WrapErr, bail};
 use prometheus::Registry;
 
+use super::RedisPool;
 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);
@@ -27,9 +27,7 @@ enum RedisBlockingPoolInner {
 }
 
 impl RedisBlockingPool {
-    pub(super) async fn new(
+    pub(super) async fn new(config: &RedisConfig) -> Result<Self> {
-        config: &RedisConfig,
-    ) -> Result<Self, RedisBackendBuildError> {
         let pool_size = config.blocking_pool_size();
         let inner = match config.topology() {
             RedisTopology::Standalone => {
@@ -39,14 +37,16 @@ impl RedisBlockingPool {
                 let manager = deadpool_redis::Manager::new_with_config(
                     config.seed_urls()[0].clone(),
                     connection_config,
-                )?;
+                )
+                .wrap_err("configuring standalone blocking Redis client")?;
                 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()?;
+                    .build()
+                    .wrap_err("building standalone blocking Redis pool")?;
                 retain_standalone_pool(pool.clone());
                 RedisBlockingPoolInner::Standalone(pool)
             }
@@ -54,14 +54,16 @@ impl RedisBlockingPool {
                 let manager = deadpool_redis::cluster::Manager::new(
                     config.seed_urls().to_vec(),
                     false,
-                )?;
+                )
+                .wrap_err("configuring clustered blocking Redis client")?;
                 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()?;
+                    .build()
+                    .wrap_err("building clustered blocking Redis pool")?;
                 retain_cluster_pool(pool.clone());
                 RedisBlockingPoolInner::Cluster(pool)
             }
@@ -70,10 +72,7 @@ impl RedisBlockingPool {
         Ok(Self { inner })
     }
 
-    pub(super) fn register_metrics(
+    pub(super) fn register_metrics(&self, registry: &Registry) -> Result<()> {
-        &self,
-        registry: &Registry,
-    ) -> Result<(), prometheus::Error> {
         register_blocking_pool_metrics(registry, self.clone())
     }
 
@@ -81,22 +80,33 @@ impl RedisBlockingPool {
         &self,
         key: &str,
         timeout: Duration,
-    ) -> Result<Option<[Vec<u8>; 2]>, Error> {
+    ) -> Result<Option<[Vec<u8>; 2]>> {
         if timeout.is_zero() {
-            return Err(Error::InvalidBlockingTimeout);
+            bail!("redis blocking timeout must be greater than zero");
         }
 
         let mut command = redis::cmd("BRPOP");
         command.arg(key).arg(timeout.as_secs_f64());
 
-        let response: Option<(Vec<u8>, Vec<u8>)> = match &self.inner {
+        let response: Option<(Vec<u8>, Vec<u8>)> =
-            RedisBlockingPoolInner::Standalone(pool) => {
+            match &self.inner {
-                command.query_async(&mut pool.get().await?).await?
+                RedisBlockingPoolInner::Standalone(pool) => {
-            }
+                    let mut connection = pool.get().await.wrap_err(
-            RedisBlockingPoolInner::Cluster(pool) => {
+                        "fetching standalone blocking Redis connection",
-                command.query_async(&mut pool.get().await?).await?
+                    )?;
-            }
+                    command.query_async(&mut connection).await.wrap_err(
-        };
+                        "reading from standalone Redis blocking queue",
+                    )?
+                }
+                RedisBlockingPoolInner::Cluster(pool) => {
+                    let mut connection = pool.get().await.wrap_err(
+                        "fetching clustered blocking Redis connection",
+                    )?;
+                    command.query_async(&mut connection).await.wrap_err(
+                        "reading from clustered Redis blocking queue",
+                    )?
+                }
+            };
 
         Ok(response.map(|(key, value)| [key, value]))
     }
@@ -120,7 +130,7 @@ impl RedisPool {
         &self,
         key: &str,
         timeout: Duration,
-    ) -> Result<Option<[Vec<u8>; 2]>, Error> {
+    ) -> Result<Option<[Vec<u8>; 2]>> {
         self.blocking.brpop(key, timeout).await
     }
 }

packages/xredis/src/cache.rs

@@ -8,6 +8,7 @@ use std::str::FromStr;
 use ariadne::ids::base62_impl::{parse_base62, to_base62};
 use chrono::{TimeZone, Utc};
 use dashmap::DashMap;
+use eyre::{Result, WrapErr, eyre};
 use futures::stream::{FuturesUnordered, StreamExt};
 use redis::aio::ConnectionLike;
 use serde::de::DeserializeOwned;
@@ -16,8 +17,6 @@ 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;
@@ -33,9 +32,7 @@ const FILL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
 pub(super) trait ConnectionProvider {
     type Connection: ConnectionLike + RoutableConnection;
 
-    fn connect(
+    fn connect(&self) -> impl Future<Output = Result<Self::Connection>> + Send;
-        &self,
-    ) -> impl Future<Output = Result<Self::Connection, Error>> + Send;
 }
 
 #[derive(Clone, Copy)]
@@ -53,7 +50,7 @@ pub enum Codec {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum EncodingFormat {
-    Json,
+    Postcard,
 }
 
 #[derive(Debug, Error)]
@@ -92,7 +89,7 @@ impl FromStr for EncodingFormat {
 
     fn from_str(value: &str) -> Result<Self, Self::Err> {
         match value {
-            "json" => Ok(Self::Json),
+            "postcard" => Ok(Self::Postcard),
             _ => Err(InvalidEncodingFormat),
         }
     }
@@ -112,12 +109,10 @@ pub struct CacheSettings {
 }
 
 impl CacheSettings {
-    pub fn encode_value<T: Serialize>(
+    pub fn encode_value<T: Serialize>(&self, value: &T) -> Result<Vec<u8>> {
-        &self,
-        value: &T,
-    ) -> Result<Vec<u8>, Error> {
         let mut value = match self.encoding_format {
-            EncodingFormat::Json => serde_json::to_vec(value)?,
+            EncodingFormat::Postcard => postcard::to_allocvec(value)
+                .wrap_err("serializing Redis cache value with postcard")?,
         };
 
         if self.compression_level > 0
@@ -148,16 +143,26 @@ impl CacheSettings {
     where
         T: for<'a> Deserialize<'a>,
     {
-        let (codec, value) = value.split_first()?;
+        let Some((codec, value)) = value.split_first() else {
-        let value = match Codec::try_from(*codec).ok()? {
+            return None;
+        };
+        let Ok(codec) = Codec::try_from(*codec) else {
+            return None;
+        };
+        let value = match codec {
             Codec::Raw => Cow::Borrowed(value),
-            Codec::Lz4 => Cow::Owned(
+            Codec::Lz4 => {
-                lz4_flex::block::decompress_size_prepended(value).ok()?,
+                let Ok(value) =
-            ),
+                    lz4_flex::block::decompress_size_prepended(value)
+                else {
+                    return None;
+                };
+                Cow::Owned(value)
+            }
         };
 
         match self.encoding_format {
-            EncodingFormat::Json => serde_json::from_slice(&value).ok(),
+            EncodingFormat::Postcard => postcard::from_bytes(&value).ok(),
         }
     }
 
@@ -202,12 +207,12 @@ impl CacheManager {
         namespace: &str,
         keys: &[K],
         closure: F,
-    ) -> Result<Vec<T>, E>
+    ) -> Result<Vec<T>>
     where
         P: ConnectionProvider,
         F: FnOnce(Vec<K>) -> Fut,
         Fut: Future<Output = Result<DashMap<K, T>, E>>,
-        E: From<Error>,
+        E: std::error::Error + Send + Sync + 'static,
         T: Serialize + DeserializeOwned,
         K: Display
             + Hash
@@ -220,7 +225,8 @@ impl CacheManager {
     {
         Ok(self
             .get_cached_keys_raw(provider, namespace, keys, closure)
-            .await?
+            .await
+            .wrap_err("fetching Redis cache values")?
             .into_values()
             .collect())
     }
@@ -232,12 +238,12 @@ impl CacheManager {
         namespace: &str,
         keys: &[K],
         closure: F,
-    ) -> Result<HashMap<K, T>, E>
+    ) -> Result<HashMap<K, T>>
     where
         P: ConnectionProvider,
         F: FnOnce(Vec<K>) -> Fut,
         Fut: Future<Output = Result<DashMap<K, T>, E>>,
-        E: From<Error>,
+        E: std::error::Error + Send + Sync + 'static,
         T: Serialize + DeserializeOwned,
         K: Display
             + Hash
@@ -255,11 +261,16 @@ impl CacheManager {
             false,
             keys,
             |ids| async move {
-                Ok(closure(ids)
+                let values = match closure(ids).await {
-                    .await?
+                    Ok(values) => values,
-                    .into_iter()
+                    Err(error) => return Err(error),
-                    .map(|(key, value)| (key, (None::<String>, value)))
+                };
-                    .collect())
+                Ok::<_, E>(
+                    values
+                        .into_iter()
+                        .map(|(key, value)| (key, (None::<String>, value)))
+                        .collect(),
+                )
             },
         )
         .await
@@ -274,12 +285,12 @@ impl CacheManager {
         case_sensitive: bool,
         keys: &[I],
         closure: F,
-    ) -> Result<Vec<T>, E>
+    ) -> Result<Vec<T>>
     where
         P: ConnectionProvider,
         F: FnOnce(Vec<I>) -> Fut,
         Fut: Future<Output = Result<DashMap<K, (Option<S>, T)>, E>>,
-        E: From<Error>,
+        E: std::error::Error + Send + Sync + 'static,
         T: Serialize + DeserializeOwned,
         I: Display + Hash + Eq + PartialEq + Clone + Debug,
         K: Display
@@ -300,7 +311,8 @@ impl CacheManager {
                 keys,
                 closure,
             )
-            .await?
+            .await
+            .wrap_err("fetching Redis cache values by slug")?
             .into_values()
             .collect())
     }
@@ -314,12 +326,12 @@ impl CacheManager {
         case_sensitive: bool,
         keys: &[I],
         closure: F,
-    ) -> Result<HashMap<K, T>, E>
+    ) -> Result<HashMap<K, T>>
     where
         P: ConnectionProvider,
         F: FnOnce(Vec<I>) -> Fut,
         Fut: Future<Output = Result<DashMap<K, (Option<S>, T)>, E>>,
-        E: From<Error>,
+        E: std::error::Error + Send + Sync + 'static,
         T: Serialize + DeserializeOwned,
         I: Display + Hash + Eq + PartialEq + Clone + Debug,
         K: Display
@@ -360,7 +372,9 @@ impl CacheManager {
                                 })
                                 .collect::<Vec<_>>();
                             let mut connection =
-                                provider.connect().await.map_err(E::from)?;
+                                provider.connect().await.wrap_err(
+                                    "connecting to Redis for slug lookup",
+                                )?;
                             let values = match routing {
                                 CacheReadRouting::ReplicaOptional => {
                                     commands::get_many_strings(
@@ -377,8 +391,8 @@ impl CacheManager {
                                     .await
                                 }
                             }
-                            .map_err(E::from)?;
+                            .wrap_err("fetching Redis cache slug values")?;
-                            Ok::<_, E>(
+                            eyre::Ok(
                                 values
                                     .into_iter()
                                     .flatten()
@@ -386,7 +400,8 @@ impl CacheManager {
                             )
                         }
                         .instrument(info_span!("get slug ids"))
-                        .await?
+                        .await
+                        .wrap_err("resolving Redis cache slugs")?
                     } else {
                         Vec::new()
                     };
@@ -403,8 +418,10 @@ impl CacheManager {
                         .map(|key| self.key_builder.entity(namespace, key))
                         .collect::<Vec<_>>();
 
-                    let mut connection =
+                    let mut connection = provider
-                        provider.connect().await.map_err(E::from)?;
+                        .connect()
+                        .await
+                        .wrap_err("connecting to Redis for cache lookup")?;
                     let mut cached_values = HashMap::new();
                     let values = match routing {
                         CacheReadRouting::ReplicaOptional => {
@@ -415,7 +432,7 @@ impl CacheManager {
                                 .await
                         }
                     }
-                    .map_err(E::from)?;
+                    .wrap_err("fetching Redis cache values")?;
                     for value in values {
                         if let Some(value) = value.and_then(|value| {
                             self.settings
@@ -425,7 +442,7 @@ impl CacheManager {
                         }
                     }
 
-                    Ok::<_, E>((cached_values, ids))
+                    eyre::Ok((cached_values, ids))
                 }
                 .instrument(info_span!("get_cached_values_closure"))
             };
@@ -437,7 +454,9 @@ impl CacheManager {
         let deadline = Instant::now() + WAIT_TIMEOUT;
 
         let (cached_values_raw, ids) =
-            get_cached_values(ids, CacheReadRouting::ReplicaOptional).await?;
+            get_cached_values(ids, CacheReadRouting::ReplicaOptional)
+                .await
+                .wrap_err("reading Redis cache")?;
         let mut cached_values = cached_values_raw
             .into_iter()
             .filter_map(|(key, value)| {
@@ -508,7 +527,10 @@ impl CacheManager {
 
                 let values = timeout_at(fill_deadline, closure(fetch_ids))
                     .await
-                    .map_err(|_| lock_timeout_error(0, waiters.len()))??;
+                    .map_err(|_| lock_timeout_error(0, waiters.len()))
+                    .wrap_err("waiting to fill Redis cache")?;
+                let values =
+                    values.wrap_err("fetching values to fill Redis cache")?;
 
                 let mut return_values = HashMap::new();
                 let mut encoded_values = Vec::with_capacity(values.len());
@@ -520,13 +542,17 @@ impl CacheManager {
                         val: value,
                         alias: slug.clone(),
                     };
-                    let encoded =
+                    let encoded = self
-                        self.settings.encode_value(&value).map_err(E::from)?;
+                        .settings
+                        .encode_value(&value)
+                        .wrap_err("encoding Redis cache value")?;
                     encoded_values.push((key, slug, value, encoded));
                 }
 
-                let mut connection =
+                let mut connection = provider
-                    provider.connect().await.map_err(E::from)?;
+                    .connect()
+                    .await
+                    .wrap_err("connecting to Redis to fill cache")?;
                 for (key, slug, _, encoded) in &encoded_values {
                     let redis_key =
                         self.key_builder.entity(namespace, key.to_string());
@@ -537,7 +563,7 @@ impl CacheManager {
                         default_expiry,
                     )
                     .await
-                    .map_err(E::from)?;
+                    .wrap_err("writing Redis cache value")?;
                     if let Some(slug) = slug
                         && let Some(slug_namespace) = slug_namespace
                     {
@@ -554,7 +580,7 @@ impl CacheManager {
                             default_expiry,
                         )
                         .await
-                        .map_err(E::from)?;
+                        .wrap_err("writing Redis cache slug")?;
                     }
                 }
 
@@ -563,7 +589,7 @@ impl CacheManager {
                     return_values.insert(key, value);
                 }
 
-                Result::<_, E>::Ok(return_values)
+                Result::<_>::Ok(return_values)
             }
             .await
         } else {
@@ -604,13 +630,14 @@ impl CacheManager {
                                 Err(error) => Err(error),
                             }
                         }
-                        Err(error) => Err(E::from(error)),
+                        Err(error) => Err(error),
                     }
                 }
             }
             Err(error) => Err(error),
         };
-        cached_values.extend(operation_result?);
+        cached_values
+            .extend(operation_result.wrap_err("populating Redis cache")?);
 
         Ok(cached_values
             .into_iter()
@@ -679,7 +706,7 @@ fn push_identity(identities: &mut Vec<String>, identity: String) {
 async fn wait_for_locks<I>(
     waiters: Vec<(I, LockWaiter)>,
     deadline: Instant,
-) -> Result<Vec<I>, Error> {
+) -> Result<Vec<I>> {
     let total = waiters.len();
     let mut released = Vec::with_capacity(total);
     let mut futures = FuturesUnordered::new();
@@ -695,26 +722,24 @@ async fn wait_for_locks<I>(
             Ok(()) => {
                 released.push(key);
             }
-            Err(error)
+            Err(_) if Instant::now() >= deadline => {
-                if is_lock_timeout(&error) || Instant::now() >= deadline =>
-            {
                 return Err(lock_timeout_error(released.len(), total));
             }
-            Err(error) => return Err(error),
+            Err(error) => {
+                return Err(error).wrap_err("waiting for Redis cache lock");
+            }
         }
     }
     Ok(released)
 }
 
-fn is_lock_timeout(error: &Error) -> bool {
+fn lock_timeout_error(
-    matches!(error, Error::LocalCacheTimeout { .. })
+    locks_released: usize,
-}
+    locks_waiting: usize,
-
+) -> eyre::Report {
-fn lock_timeout_error(locks_released: usize, locks_waiting: usize) -> Error {
+    eyre!(
-    Error::LocalCacheTimeout {
+        "timeout waiting on local Redis cache lock ({locks_released}/{locks_waiting} released)"
-        released: locks_released,
+    )
-        total: locks_waiting,
-    }
 }
 
 #[derive(Serialize, Deserialize)]

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

@@ -3,11 +3,10 @@ use std::sync::atomic::{AtomicBool, Ordering};
 
 use dashmap::DashMap;
 use dashmap::mapref::entry::Entry;
+use eyre::{Result, WrapErr};
 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>>>,
@@ -76,10 +75,7 @@ pub(in crate::cache) struct LockWaiter {
 }
 
 impl LockWaiter {
-    pub(in crate::cache) async fn wait(
+    pub(in crate::cache) async fn wait(self, deadline: Instant) -> Result<()> {
-        self,
-        deadline: Instant,
-    ) -> Result<(), Error> {
         loop {
             if self.state.released.load(Ordering::Acquire) {
                 return Ok(());
@@ -94,7 +90,7 @@ impl LockWaiter {
 
             timeout_at(deadline, notified)
                 .await
-                .map_err(|_| lock_timeout())?;
+                .wrap_err("waiting for local Redis cache lock")?;
         }
     }
 }
@@ -112,10 +108,3 @@ impl LockState {
         }
     }
 }
-
-fn lock_timeout() -> Error {
-    Error::LocalCacheTimeout {
-        released: 0,
-        total: 1,
-    }
-}

packages/xredis/src/commands.rs

@@ -1,10 +1,9 @@
 use std::fmt::Debug;
 
+use eyre::{Result, WrapErr};
 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;
@@ -18,7 +17,7 @@ pub async fn set<C, D>(
     key: &str,
     data: D,
     expiry: i64,
-) -> Result<(), Error>
+) -> Result<()>
 where
     C: ConnectionLike,
     D: ToRedisArgs + Send + Sync + Debug,
@@ -29,7 +28,8 @@ where
         .arg("EX")
         .arg(expiry)
         .query_async::<()>(connection)
-        .await?;
+        .await
+        .wrap_err("writing to Redis")?;
     Ok(())
 }
 
@@ -40,7 +40,7 @@ pub async fn set_serialized<C, D>(
     data: D,
     expiry: Option<i64>,
     settings: &CacheSettings,
-) -> Result<(), Error>
+) -> Result<()>
 where
     C: ConnectionLike,
     D: serde::Serialize,
@@ -48,21 +48,24 @@ where
     set(
         connection,
         key,
-        settings.encode_value(&data)?,
+        settings
+            .encode_value(&data)
+            .wrap_err("serializing Redis value")?,
         expiry.unwrap_or(settings.default_expiry),
     )
     .await
 }
 
 #[tracing::instrument(skip_all)]
-pub async fn get<C>(
+pub async fn get<C>(connection: &mut C, key: &str) -> Result<Option<String>>
-    connection: &mut C,
-    key: &str,
-) -> Result<Option<String>, Error>
 where
     C: ConnectionLike,
 {
-    Ok(cmd("GET").arg(key).query_async(connection).await?)
+    cmd("GET")
+        .arg(key)
+        .query_async(connection)
+        .await
+        .wrap_err("fetching from Redis")
 }
 
 /// Issues ordinary `MGET` commands in bounded chunks. Cluster routing and
@@ -72,7 +75,7 @@ where
 pub async fn get_many<C>(
     connection: &mut C,
     keys: &[String],
-) -> Result<Vec<Option<Vec<u8>>>, Error>
+) -> Result<Vec<Option<Vec<u8>>>>
 where
     C: ConnectionLike,
 {
@@ -83,7 +86,7 @@ where
 pub async fn get_many_strings<C>(
     connection: &mut C,
     keys: &[String],
-) -> Result<Vec<Option<String>>, Error>
+) -> Result<Vec<Option<String>>>
 where
     C: ConnectionLike,
 {
@@ -93,7 +96,7 @@ where
 pub(super) async fn get_many_primary<C>(
     connection: &mut C,
     keys: &[String],
-) -> Result<Vec<Option<Vec<u8>>>, Error>
+) -> Result<Vec<Option<Vec<u8>>>>
 where
     C: RoutableConnection,
 {
@@ -103,7 +106,7 @@ where
 pub(super) async fn get_many_strings_primary<C>(
     connection: &mut C,
     keys: &[String],
-) -> Result<Vec<Option<String>>, Error>
+) -> Result<Vec<Option<String>>>
 where
     C: RoutableConnection,
 {
@@ -113,7 +116,7 @@ where
 pub(super) async fn get_many_as<C, T>(
     connection: &mut C,
     keys: &[String],
-) -> Result<Vec<Option<T>>, Error>
+) -> Result<Vec<Option<T>>>
 where
     C: ConnectionLike,
     T: FromRedisValue,
@@ -123,7 +126,8 @@ where
         let part = cmd("MGET")
             .arg(chunk)
             .query_async::<Vec<Option<T>>>(connection)
-            .await?;
+            .await
+            .wrap_err("fetching multiple values from Redis")?;
         values.extend(part);
     }
     Ok(values)
@@ -132,7 +136,7 @@ where
 async fn get_many_primary_as<C, T>(
     connection: &mut C,
     keys: &[String],
-) -> Result<Vec<Option<T>>, Error>
+) -> Result<Vec<Option<T>>>
 where
     C: RoutableConnection,
     T: FromRedisValue,
@@ -143,10 +147,14 @@ where
         command.arg(chunk);
         let value = connection
             .route_command(command, primary_mget_routing(chunk))
-            .await?;
+            .await
-        let part =
+            .wrap_err("fetching multiple values from primary Redis nodes")?;
-            redis::from_redis_value::<Vec<Option<T>>>(value.extract_error()?)
+        let value = value
-                .map_err(redis::RedisError::from)?;
+            .extract_error()
+            .wrap_err("extracting Redis response")?;
+        let part = redis::from_redis_value::<Vec<Option<T>>>(value)
+            .map_err(redis::RedisError::from)
+            .wrap_err("decoding Redis response")?;
         values.extend(part);
     }
     Ok(values)
@@ -157,13 +165,16 @@ pub async fn get_deserialized<C, R>(
     connection: &mut C,
     key: &str,
     settings: &CacheSettings,
-) -> Result<Option<R>, Error>
+) -> Result<Option<R>>
 where
     C: ConnectionLike,
     R: for<'a> serde::Deserialize<'a>,
 {
-    let value: Option<Vec<u8>> =
+    let value: Option<Vec<u8>> = cmd("GET")
-        cmd("GET").arg(key).query_async(connection).await?;
+        .arg(key)
+        .query_async(connection)
+        .await
+        .wrap_err("fetching serialized value from Redis")?;
     Ok(value.and_then(|value| settings.decode_value(&value)))
 }
 
@@ -172,47 +183,49 @@ pub async fn get_many_deserialized<C, R>(
     connection: &mut C,
     keys: &[String],
     settings: &CacheSettings,
-) -> Result<Vec<Option<R>>, Error>
+) -> Result<Vec<Option<R>>>
 where
     C: ConnectionLike,
     R: for<'a> serde::Deserialize<'a>,
 {
     Ok(get_many(connection, keys)
-        .await?
+        .await
+        .wrap_err("fetching serialized values from Redis")?
         .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>
+pub async fn delete<C>(connection: &mut C, key: &str) -> Result<()>
 where
     C: ConnectionLike,
 {
-    cmd("DEL").arg(key).query_async::<()>(connection).await?;
+    cmd("DEL")
+        .arg(key)
+        .query_async::<()>(connection)
+        .await
+        .wrap_err("deleting from Redis")?;
     Ok(())
 }
 
 #[tracing::instrument(skip_all)]
-pub async fn delete_many<C>(
+pub async fn delete_many<C>(connection: &mut C, keys: &[String]) -> Result<()>
-    connection: &mut C,
-    keys: &[String],
-) -> Result<(), Error>
 where
     C: ConnectionLike,
 {
     if !keys.is_empty() {
-        cmd("DEL").arg(keys).query_async::<()>(connection).await?;
+        cmd("DEL")
+            .arg(keys)
+            .query_async::<()>(connection)
+            .await
+            .wrap_err("deleting multiple values from Redis")?;
     }
     Ok(())
 }
 
 #[tracing::instrument(skip_all)]
-pub async fn lpush<C, D>(
+pub async fn lpush<C, D>(connection: &mut C, key: &str, value: D) -> Result<()>
-    connection: &mut C,
-    key: &str,
-    value: D,
-) -> Result<(), Error>
 where
     C: ConnectionLike,
     D: ToRedisArgs + Send + Sync + Debug,
@@ -221,17 +234,19 @@ where
         .arg(key)
         .arg(value)
         .query_async::<()>(connection)
-        .await?;
+        .await
+        .wrap_err("pushing to Redis list")?;
     Ok(())
 }
 
 #[tracing::instrument(skip_all)]
-pub async fn incr<C>(
+pub async fn incr<C>(connection: &mut C, key: &str) -> Result<Option<u64>>
-    connection: &mut C,
-    key: &str,
-) -> Result<Option<u64>, Error>
 where
     C: ConnectionLike,
 {
-    Ok(cmd("INCR").arg(key).query_async(connection).await?)
+    cmd("INCR")
+        .arg(key)
+        .query_async(connection)
+        .await
+        .wrap_err("incrementing Redis value")
 }

packages/xredis/src/config.rs

@@ -1,5 +1,6 @@
 use std::{fmt, str::FromStr};
 
+use eyre::{Result, WrapErr};
 use thiserror::Error;
 
 #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
@@ -91,13 +92,11 @@ pub(crate) struct RedisPoolSize {
 }
 
 impl RedisPoolSize {
-    fn new(
+    fn new(name: &'static str, max: usize, min: usize) -> Result<Self> {
-        name: &'static str,
-        max: usize,
-        min: usize,
-    ) -> Result<Self, RedisConfigError> {
         if max == 0 || min > max {
-            return Err(RedisConfigError::InvalidPoolSize { name, max, min });
+            return Err(
+                RedisConfigError::InvalidPoolSize { name, max, min }.into()
+            );
         }
 
         Ok(Self { max, min })
@@ -193,11 +192,12 @@ impl RedisConfig {
         blocking_pool_size: (usize, usize),
         cache_locking_strategy: CacheLockingStrategy,
         read_replica_strategy: ReadReplicaStrategy,
-    ) -> Result<Self, RedisConfigError> {
+    ) -> Result<Self> {
         if cache_locking_strategy == CacheLockingStrategy::Distributed {
             return Err(RedisConfigError::UnsupportedCacheLockingStrategy {
                 strategy: cache_locking_strategy,
-            });
+            }
+            .into());
         }
 
         let seed_urls = raw_urls
@@ -208,26 +208,32 @@ impl RedisConfig {
             .collect::<Vec<_>>();
 
         if seed_urls.is_empty() {
-            return Err(RedisConfigError::MissingUrl);
+            return Err(RedisConfigError::MissingUrl.into());
         }
 
         let backend = match (mode, connection_type) {
             (RedisTopology::Standalone, RedisConnectionType::Pooled) => {
                 if seed_urls.len() != 1 {
-                    return Err(RedisConfigError::MultipleStandaloneUrls);
+                    return Err(RedisConfigError::MultipleStandaloneUrls.into());
                 }
-                RedisBackendConfig::StandalonePooled(RedisPoolSize::new(
+                RedisBackendConfig::StandalonePooled(
-                    "standalone",
+                    RedisPoolSize::new(
-                    standalone_pool_size.0,
+                        "standalone",
-                    standalone_pool_size.1,
+                        standalone_pool_size.0,
-                )?)
+                        standalone_pool_size.1,
+                    )
+                    .wrap_err("validating standalone Redis pool size")?,
+                )
             }
             (RedisTopology::Cluster, RedisConnectionType::Pooled) => {
-                RedisBackendConfig::ClusterPooled(RedisPoolSize::new(
+                RedisBackendConfig::ClusterPooled(
-                    "cluster",
+                    RedisPoolSize::new(
-                    cluster_pool_size.0,
+                        "cluster",
-                    cluster_pool_size.1,
+                        cluster_pool_size.0,
-                )?)
+                        cluster_pool_size.1,
+                    )
+                    .wrap_err("validating clustered Redis pool size")?,
+                )
             }
             (RedisTopology::Cluster, RedisConnectionType::Multiplexed) => {
                 RedisBackendConfig::ClusterMultiplexed
@@ -236,7 +242,8 @@ impl RedisConfig {
                 return Err(RedisConfigError::UnsupportedConnectionType {
                     mode,
                     connection_type,
-                });
+                }
+                .into());
             }
         };
 
@@ -249,7 +256,8 @@ impl RedisConfig {
                 "blocking",
                 blocking_pool_size.0,
                 blocking_pool_size.1,
-            )?,
+            )
+            .wrap_err("validating blocking Redis pool size")?,
             cache_locking_strategy,
             read_replica_strategy,
         })

packages/xredis/src/connection.rs

@@ -1,5 +1,6 @@
 use std::time::Duration;
 
+use eyre::{Result, WrapErr};
 use futures::future::try_join_all;
 use prometheus::Registry;
 use redis::aio::ConnectionLike;
@@ -7,7 +8,6 @@ use redis::cluster_read_routing::{
     RandomReplicaStrategy, RoundRobinReplicaStrategy,
 };
 use redis::cluster_routing::RoutingInfo;
-use thiserror::Error;
 use tracing::warn;
 
 use crate::ReadReplicaStrategy;
@@ -29,16 +29,6 @@ pub(crate) enum RedisBackend {
     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,
 }
@@ -58,9 +48,7 @@ pub(crate) trait RoutableConnection: ConnectionLike {
 }
 
 impl RedisBackend {
-    pub(crate) async fn new(
+    pub(crate) async fn new(config: &RedisConfig) -> Result<Self> {
-        config: &RedisConfig,
-    ) -> Result<Self, RedisBackendBuildError> {
         match config.backend() {
             RedisBackendConfig::StandalonePooled(pool_size) => {
                 Self::standalone_pooled(config, pool_size).await
@@ -77,21 +65,25 @@ impl RedisBackend {
     async fn standalone_pooled(
         config: &RedisConfig,
         pool_size: RedisPoolSize,
-    ) -> Result<Self, RedisBackendBuildError> {
+    ) -> Result<Self> {
         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,
-        )?;
+        )
+        .wrap_err("configuring standalone Redis client")?;
         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()?;
+            .build()
+            .wrap_err("building standalone Redis pool")?;
 
-        warm_standalone_pool(&pool, pool_size.min()).await?;
+        warm_standalone_pool(&pool, pool_size.min())
+            .await
+            .wrap_err("warming standalone Redis pool")?;
         retain_standalone_pool(pool.clone());
 
         Ok(Self::StandalonePooled(pool))
@@ -100,16 +92,18 @@ impl RedisBackend {
     async fn cluster_pooled(
         config: &RedisConfig,
         pool_size: RedisPoolSize,
-    ) -> Result<Self, RedisBackendBuildError> {
+    ) -> Result<Self> {
         let manager = deadpool_redis::cluster::Manager::new(
             config.seed_urls().to_vec(),
             false,
-        )?;
+        )
+        .wrap_err("configuring clustered Redis client")?;
         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()?;
+            .build()
+            .wrap_err("building clustered Redis pool")?;
 
         if config.read_replica_strategy() != ReadReplicaStrategy::Primary {
             warn!(
@@ -117,15 +111,15 @@ impl RedisBackend {
             );
         }
 
-        warm_cluster_pool(&pool, pool_size.min()).await?;
+        warm_cluster_pool(&pool, pool_size.min())
+            .await
+            .wrap_err("warming clustered Redis pool")?;
         retain_cluster_pool(pool.clone());
 
         Ok(Self::ClusterPooled(pool))
     }
 
-    async fn cluster_multiplexed(
+    async fn cluster_multiplexed(config: &RedisConfig) -> Result<Self> {
-        config: &RedisConfig,
-    ) -> Result<Self, RedisBackendBuildError> {
         let mut builder = redis::cluster::ClusterClientBuilder::new(
             config.seed_urls().iter().map(String::as_str),
         );
@@ -141,22 +135,31 @@ impl RedisBackend {
             }
         }
 
-        let client = builder.build()?;
+        let client = builder
-        let connection = client.get_async_connection().await?;
+            .build()
+            .wrap_err("building multiplexed Redis client")?;
+        let connection = client
+            .get_async_connection()
+            .await
+            .wrap_err("connecting multiplexed Redis client")?;
 
         Ok(Self::ClusterMultiplexed(connection))
     }
 
-    pub(crate) async fn connect(
+    pub(crate) async fn connect(&self) -> Result<RedisConnection> {
-        &self,
-    ) -> Result<RedisConnection, deadpool_redis::PoolError> {
         let inner = match self {
             Self::StandalonePooled(pool) => {
-                RedisConnectionInner::StandalonePooled(pool.get().await?)
+                RedisConnectionInner::StandalonePooled(
-            }
+                    pool.get()
-            Self::ClusterPooled(pool) => {
+                        .await
-                RedisConnectionInner::ClusterPooled(pool.get().await?)
+                        .wrap_err("fetching standalone Redis connection")?,
+                )
             }
+            Self::ClusterPooled(pool) => RedisConnectionInner::ClusterPooled(
+                pool.get()
+                    .await
+                    .wrap_err("fetching clustered Redis connection")?,
+            ),
             Self::ClusterMultiplexed(connection) => {
                 RedisConnectionInner::ClusterMultiplexed(connection.clone())
             }
@@ -165,10 +168,7 @@ impl RedisBackend {
         Ok(RedisConnection { inner })
     }
 
-    pub(crate) fn register_metrics(
+    pub(crate) fn register_metrics(&self, registry: &Registry) -> Result<()> {
-        &self,
-        registry: &Registry,
-    ) -> Result<(), prometheus::Error> {
         register_command_pool_metrics(registry, self.clone())
     }
 }
@@ -282,8 +282,10 @@ impl RoutableConnection for RedisConnection {
 async fn warm_standalone_pool(
     pool: &deadpool_redis::Pool,
     min: usize,
-) -> Result<(), deadpool_redis::PoolError> {
+) -> Result<()> {
-    let connections = try_join_all((0..min).map(|_| pool.get())).await?;
+    let connections = try_join_all((0..min).map(|_| pool.get()))
+        .await
+        .wrap_err("fetching initial standalone Redis connections")?;
     drop(connections);
     Ok(())
 }
@@ -291,8 +293,10 @@ async fn warm_standalone_pool(
 async fn warm_cluster_pool(
     pool: &deadpool_redis::cluster::Pool,
     min: usize,
-) -> Result<(), deadpool_redis::PoolError> {
+) -> Result<()> {
-    let connections = try_join_all((0..min).map(|_| pool.get())).await?;
+    let connections = try_join_all((0..min).map(|_| pool.get()))
+        .await
+        .wrap_err("fetching initial clustered Redis connections")?;
     drop(connections);
     Ok(())
 }

packages/xredis/src/lib.rs

@@ -6,6 +6,7 @@ use std::hash::Hash;
 use std::sync::Arc;
 
 use dashmap::DashMap;
+use eyre::{Result, WrapErr};
 use prometheus::Registry;
 use redis::aio::ConnectionLike;
 use redis::{FromRedisValue, ToRedisArgs};
@@ -35,25 +36,7 @@ pub use config::{
     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 {
@@ -75,15 +58,19 @@ impl RedisPool {
         meta_namespace: impl Into<Arc<str>>,
         config: RedisConfig,
         cache_settings: CacheSettings,
-    ) -> Result<Self, RedisBackendBuildError> {
+    ) -> Result<Self> {
         tracing::info!(
             strategy = %config.cache_locking_strategy(),
             "configured Redis cache locking"
         );
 
-        let backend = RedisBackend::new(&config).await?;
+        let backend = RedisBackend::new(&config)
+            .await
+            .wrap_err("creating Redis command backend")?;
 
-        let blocking = blocking::RedisBlockingPool::new(&config).await?;
+        let blocking = blocking::RedisBlockingPool::new(&config)
+            .await
+            .wrap_err("creating Redis blocking pool")?;
         let key_builder = KeyBuilder::new(meta_namespace, config.topology());
         let cache = CacheManager::new(key_builder.clone(), cache_settings);
 
@@ -102,9 +89,13 @@ impl RedisPool {
 }
 
 impl RedisPool {
-    pub async fn connect(&self) -> Result<RedisConnection, Error> {
+    pub async fn connect(&self) -> Result<RedisConnection> {
         Ok(RedisConnection {
-            inner: self.backend.connect().await?,
+            inner: self
+                .backend
+                .connect()
+                .await
+                .wrap_err("connecting to Redis")?,
             key_builder: self.key_builder.clone(),
             settings: self.cache.settings().clone(),
         })
@@ -113,9 +104,13 @@ impl RedisPool {
     pub async fn register_and_set_metrics(
         &self,
         registry: &Registry,
-    ) -> Result<(), prometheus::Error> {
+    ) -> Result<()> {
-        self.backend.register_metrics(registry)?;
+        self.backend
-        self.blocking.register_metrics(registry)
+            .register_metrics(registry)
+            .wrap_err("registering Redis command pool metrics")?;
+        self.blocking
+            .register_metrics(registry)
+            .wrap_err("registering Redis blocking pool metrics")
     }
 
     pub async fn get_cached_keys<F, Fut, T, K, E>(
@@ -123,11 +118,11 @@ impl RedisPool {
         namespace: &str,
         keys: &[K],
         closure: F,
-    ) -> Result<Vec<T>, E>
+    ) -> Result<Vec<T>>
     where
         F: FnOnce(Vec<K>) -> Fut,
         Fut: Future<Output = Result<DashMap<K, T>, E>>,
-        E: From<Error>,
+        E: std::error::Error + Send + Sync + 'static,
         T: Serialize + DeserializeOwned,
         K: Display
             + Hash
@@ -148,11 +143,11 @@ impl RedisPool {
         namespace: &str,
         keys: &[K],
         closure: F,
-    ) -> Result<std::collections::HashMap<K, T>, E>
+    ) -> Result<std::collections::HashMap<K, T>>
     where
         F: FnOnce(Vec<K>) -> Fut,
         Fut: Future<Output = Result<DashMap<K, T>, E>>,
-        E: From<Error>,
+        E: std::error::Error + Send + Sync + 'static,
         T: Serialize + DeserializeOwned,
         K: Display
             + Hash
@@ -175,11 +170,11 @@ impl RedisPool {
         case_sensitive: bool,
         keys: &[I],
         closure: F,
-    ) -> Result<Vec<T>, E>
+    ) -> Result<Vec<T>>
     where
         F: FnOnce(Vec<I>) -> Fut,
         Fut: Future<Output = Result<DashMap<K, (Option<S>, T)>, E>>,
-        E: From<Error>,
+        E: std::error::Error + Send + Sync + 'static,
         T: Serialize + DeserializeOwned,
         I: Display + Hash + Eq + PartialEq + Clone + Debug,
         K: Display
@@ -210,11 +205,11 @@ impl RedisPool {
         case_sensitive: bool,
         keys: &[I],
         closure: F,
-    ) -> Result<std::collections::HashMap<K, T>, E>
+    ) -> Result<std::collections::HashMap<K, T>>
     where
         F: FnOnce(Vec<I>) -> Fut,
         Fut: Future<Output = Result<DashMap<K, (Option<S>, T)>, E>>,
-        E: From<Error>,
+        E: std::error::Error + Send + Sync + 'static,
         T: Serialize + DeserializeOwned,
         I: Display + Hash + Eq + PartialEq + Clone + Debug,
         K: Display
@@ -242,9 +237,7 @@ impl RedisPool {
 impl ConnectionProvider for RedisPool {
     type Connection = RedisConnection;
 
-    fn connect(
+    fn connect(&self) -> impl Future<Output = Result<Self::Connection>> + Send {
-        &self,
-    ) -> impl Future<Output = Result<Self::Connection, Error>> + Send {
         RedisPool::connect(self)
     }
 }
@@ -259,7 +252,7 @@ impl RedisConnection {
         key: &str,
         data: D,
         expiry: Option<i64>,
-    ) -> Result<(), Error>
+    ) -> Result<()>
     where
         D: ToRedisArgs + Send + Sync + Debug,
     {
@@ -277,7 +270,7 @@ impl RedisConnection {
         key: &str,
         data: D,
         expiry: Option<i64>,
-    ) -> Result<(), Error>
+    ) -> Result<()>
     where
         D: Serialize,
     {
@@ -291,31 +284,28 @@ impl RedisConnection {
         .await
     }
 
-    pub async fn get(&mut self, key: &str) -> Result<Option<String>, Error> {
+    pub async fn get(&mut self, key: &str) -> Result<Option<String>> {
         commands::get(&mut self.inner, key).await
     }
 
     pub async fn get_many(
         &mut self,
         keys: &[String],
-    ) -> Result<Vec<Option<Vec<u8>>>, Error> {
+    ) -> Result<Vec<Option<Vec<u8>>>> {
         commands::get_many(&mut self.inner, keys).await
     }
 
     pub async fn get_many_typed<R>(
         &mut self,
         keys: &[String],
-    ) -> Result<Vec<Option<R>>, Error>
+    ) -> Result<Vec<Option<R>>>
     where
         R: FromRedisValue,
     {
         commands::get_many_as(&mut self.inner, keys).await
     }
 
-    pub async fn get_deserialized<R>(
+    pub async fn get_deserialized<R>(&mut self, key: &str) -> Result<Option<R>>
-        &mut self,
-        key: &str,
-    ) -> Result<Option<R>, Error>
     where
         R: for<'a> serde::Deserialize<'a>,
     {
@@ -325,7 +315,7 @@ impl RedisConnection {
     pub async fn get_many_deserialized<R>(
         &mut self,
         keys: &[String],
-    ) -> Result<Vec<Option<R>>, Error>
+    ) -> Result<Vec<Option<R>>>
     where
         R: for<'a> serde::Deserialize<'a>,
     {
@@ -333,22 +323,22 @@ impl RedisConnection {
             .await
     }
 
-    pub async fn delete(&mut self, key: &str) -> Result<(), Error> {
+    pub async fn delete(&mut self, key: &str) -> Result<()> {
         commands::delete(&mut self.inner, key).await
     }
 
-    pub async fn delete_many(&mut self, keys: &[String]) -> Result<(), Error> {
+    pub async fn delete_many(&mut self, keys: &[String]) -> Result<()> {
         commands::delete_many(&mut self.inner, keys).await
     }
 
-    pub async fn lpush<D>(&mut self, key: &str, value: D) -> Result<(), Error>
+    pub async fn lpush<D>(&mut self, key: &str, value: D) -> Result<()>
     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> {
+    pub async fn incr(&mut self, key: &str) -> Result<Option<u64>> {
         commands::incr(&mut self.inner, key).await
     }
 }

packages/xredis/src/metrics.rs

@@ -1,5 +1,6 @@
 use std::time::Duration;
 
+use eyre::{Result, WrapErr};
 use prometheus::{IntGauge, Registry};
 
 const METRICS_UPDATE_INTERVAL: Duration = Duration::from_secs(5);
@@ -72,7 +73,7 @@ impl RedisPoolMetrics {
     fn register(
         registry: &Registry,
         kind: RedisPoolMetricsKind,
-    ) -> Result<Self, prometheus::Error> {
+    ) -> Result<Self> {
         let prefix = kind.metric_prefix();
         let description = kind.description();
         let max_size = IntGauge::new(
@@ -80,28 +81,40 @@ impl RedisPoolMetrics {
             format!(
                 "Maximum logical connection count for the {description}; clustered logical connections may own multiple physical sockets"
             ),
-        )?;
+        )
+        .wrap_err("creating Redis pool maximum size metric")?;
         let size = IntGauge::new(
             format!("{prefix}_size"),
             format!(
                 "Current logical connection count for the {description}; clustered logical connections may own multiple physical sockets"
             ),
-        )?;
+        )
+        .wrap_err("creating Redis pool size metric")?;
         let available = IntGauge::new(
             format!("{prefix}_available"),
             format!("Available logical connections in the {description}"),
-        )?;
+        )
+        .wrap_err("creating Redis pool availability metric")?;
         let waiting = IntGauge::new(
             format!("{prefix}_waiting"),
             format!(
                 "Number of futures waiting for a logical connection from the {description}"
             ),
-        )?;
+        )
+        .wrap_err("creating Redis pool waiters metric")?;
 
-        registry.register(Box::new(max_size.clone()))?;
+        registry
-        registry.register(Box::new(size.clone()))?;
+            .register(Box::new(max_size.clone()))
-        registry.register(Box::new(available.clone()))?;
+            .wrap_err("registering Redis pool maximum size metric")?;
-        registry.register(Box::new(waiting.clone()))?;
+        registry
+            .register(Box::new(size.clone()))
+            .wrap_err("registering Redis pool size metric")?;
+        registry
+            .register(Box::new(available.clone()))
+            .wrap_err("registering Redis pool availability metric")?;
+        registry
+            .register(Box::new(waiting.clone()))
+            .wrap_err("registering Redis pool waiters metric")?;
 
         Ok(Self {
             max_size,
@@ -122,7 +135,7 @@ impl RedisPoolMetrics {
 pub(super) fn register_command_pool_metrics<P>(
     registry: &Registry,
     provider: P,
-) -> Result<(), prometheus::Error>
+) -> Result<()>
 where
     P: LogicalPoolStatusProvider,
 {
@@ -132,7 +145,7 @@ where
 pub(super) fn register_blocking_pool_metrics<P>(
     registry: &Registry,
     provider: P,
-) -> Result<(), prometheus::Error>
+) -> Result<()>
 where
     P: LogicalPoolStatusProvider,
 {
@@ -143,11 +156,12 @@ fn register_pool_metrics<P>(
     registry: &Registry,
     kind: RedisPoolMetricsKind,
     provider: P,
-) -> Result<(), prometheus::Error>
+) -> Result<()>
 where
     P: LogicalPoolStatusProvider,
 {
-    let metrics = RedisPoolMetrics::register(registry, kind)?;
+    let metrics = RedisPoolMetrics::register(registry, kind)
+        .wrap_err("registering Redis pool metrics")?;
     metrics.set(provider.logical_pool_status());
 
     tokio::spawn(async move {

packages/xredis/src/pubsub.rs

@@ -1,11 +1,12 @@
 use std::time::Duration;
 
+use eyre::{Result, WrapErr};
 use futures::StreamExt;
 use redis::ToRedisArgs;
 use tokio::sync::mpsc;
 use tracing::{info, warn};
 
-use super::{Error, RedisPool};
+use super::RedisPool;
 
 const PUBSUB_BUFFER_SIZE: usize = 1024;
 const INITIAL_RECONNECT_BACKOFF: Duration = Duration::from_millis(250);
@@ -35,21 +36,20 @@ impl RedisPool {
         receiver
     }
 
-    pub async fn publish<M>(
+    pub async fn publish<M>(&self, channel: &str, message: M) -> Result<()>
-        &self,
-        channel: &str,
-        message: M,
-    ) -> Result<(), Error>
     where
         M: ToRedisArgs + Send + Sync,
     {
-        let mut connection = self.connect().await?;
+        let mut connection = self
+            .connect()
+            .await
+            .wrap_err("connecting to Redis for publishing")?;
         let _: usize = redis::cmd("PUBLISH")
             .arg(channel)
             .arg(message)
             .query_async(&mut connection)
             .await
-            .map_err(Error::from)?;
+            .wrap_err("publishing to Redis channel")?;
         Ok(())
     }
 }
@@ -111,10 +111,17 @@ async fn forward_from_seed(
     seed_url: &str,
     channel: &'static str,
     sender: &mpsc::Sender<Vec<u8>>,
-) -> redis::RedisResult<SubscriptionOutcome> {
+) -> Result<SubscriptionOutcome> {
-    let client = redis::Client::open(seed_url)?;
+    let client = redis::Client::open(seed_url)
-    let mut pubsub = client.get_async_pubsub().await?;
+        .wrap_err("configuring Redis Pub/Sub client")?;
-    pubsub.subscribe(channel).await?;
+    let mut pubsub = client
+        .get_async_pubsub()
+        .await
+        .wrap_err("connecting to Redis Pub/Sub")?;
+    pubsub
+        .subscribe(channel)
+        .await
+        .wrap_err("subscribing to Redis channel")?;
     info!(channel, "Established Redis Pub/Sub subscription");
 
     let mut stream = pubsub.into_on_message();