diff --git a/Cargo.lock b/Cargo.lock index b3eefef689..8c0ecd0b9a 100644 --- a/Cargo.lock +++ b/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", diff --git a/apps/labrinth/AGENTS.md b/apps/labrinth/AGENTS.md index 0c1e193f6f..3c11a2248b 100644 --- a/apps/labrinth/AGENTS.md +++ b/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` instead of `eyre::Result` +- 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` diff --git a/apps/labrinth/src/auth/mod.rs b/apps/labrinth/src/auth/mod.rs index bf0a963f81..c58a032f13 100644 --- a/apps/labrinth/src/auth/mod.rs +++ b/apps/labrinth/src/auth/mod.rs @@ -59,12 +59,6 @@ pub enum AuthenticationError { Url, } -impl From 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 { diff --git a/apps/labrinth/src/database/models/mod.rs b/apps/labrinth/src/database/models/mod.rs index 09eddcaed7..badfad0bdb 100644 --- a/apps/labrinth/src/database/models/mod.rs +++ b/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), } diff --git a/apps/labrinth/src/database/models/notifications_template_item.rs b/apps/labrinth/src/database/models/notifications_template_item.rs index e5b7911561..5169e670be 100644 --- a/apps/labrinth/src/database/models/notifications_template_item.rs +++ b/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) = - redis_conn.get_deserialized::(&redis_key).await? + if let Some(body) = redis_conn + .get_deserialized::(&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) } diff --git a/apps/labrinth/src/database/models/project_item.rs b/apps/labrinth/src/database/models/project_item.rs index e862cf76f3..2b9092d489 100644 --- a/apps/labrinth/src/database/models/project_item.rs +++ b/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) } diff --git a/apps/labrinth/src/database/redis.rs b/apps/labrinth/src/database/redis.rs index 0bc5f37231..d7d54c7fc2 100644 --- a/apps/labrinth/src/database/redis.rs +++ b/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 { + fn from_env() -> Result { 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, diff --git a/apps/labrinth/src/env.rs b/apps/labrinth/src/env.rs index 88978d006a..9ddbdb9c24 100644 --- a/apps/labrinth/src/env.rs +++ b/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. diff --git a/apps/labrinth/src/queue/analytics/mod.rs b/apps/labrinth/src/queue/analytics/mod.rs index 18766e48da..6849baa63d 100644 --- a/apps/labrinth/src/queue/analytics/mod.rs +++ b/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::>(); - 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 = - redis_connection.get_many_typed::(&redis_keys).await?; + let results = redis_connection + .get_many_typed::(&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::>(); - 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 = - redis_connection.get_many_typed::(&redis_keys).await?; + let results = redis_connection + .get_many_typed::(&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::("views").await?; @@ -267,10 +280,15 @@ impl AnalyticsQueue { ) }) .collect::>(); - 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 = - redis_connection.get_many_typed::(&redis_keys).await?; + let results = redis_connection + .get_many_typed::(&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?; diff --git a/apps/labrinth/src/routes/mod.rs b/apps/labrinth/src/routes/mod.rs index fc75849d22..e9e606deba 100644 --- a/apps/labrinth/src/routes/mod.rs +++ b/apps/labrinth/src/routes/mod.rs @@ -267,12 +267,6 @@ pub enum ApiError { }, } -impl From for ApiError { - fn from(error: xredis::Error) -> Self { - Self::Database(error.into()) - } -} - impl ApiError { pub fn delphi(err: impl Into) -> Self { Self::Delphi(err.into()) diff --git a/packages/xredis/Cargo.toml b/packages/xredis/Cargo.toml index 7f9fd2035a..407f648de3 100644 --- a/packages/xredis/Cargo.toml +++ b/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 } diff --git a/packages/xredis/src/blocking.rs b/packages/xredis/src/blocking.rs index 7a7b7f4e0f..5aaf62d772 100644 --- a/packages/xredis/src/blocking.rs +++ b/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( - config: &RedisConfig, - ) -> Result { + pub(super) async fn new(config: &RedisConfig) -> Result { 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( - &self, - registry: &Registry, - ) -> Result<(), prometheus::Error> { + pub(super) fn register_metrics(&self, registry: &Registry) -> Result<()> { register_blocking_pool_metrics(registry, self.clone()) } @@ -81,22 +80,33 @@ impl RedisBlockingPool { &self, key: &str, timeout: Duration, - ) -> Result; 2]>, Error> { + ) -> Result; 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, Vec)> = 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? - } - }; + let response: Option<(Vec, Vec)> = + match &self.inner { + RedisBlockingPoolInner::Standalone(pool) => { + let mut connection = pool.get().await.wrap_err( + "fetching standalone blocking Redis connection", + )?; + 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; 2]>, Error> { + ) -> Result; 2]>> { self.blocking.brpop(key, timeout).await } } diff --git a/packages/xredis/src/cache.rs b/packages/xredis/src/cache.rs index 5e2aba678b..d14fa61356 100644 --- a/packages/xredis/src/cache.rs +++ b/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( - &self, - ) -> impl Future> + Send; + fn connect(&self) -> impl Future> + 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 { 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( - &self, - value: &T, - ) -> Result, Error> { + pub fn encode_value(&self, value: &T) -> Result> { 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 value = match Codec::try_from(*codec).ok()? { + let Some((codec, value)) = value.split_first() else { + 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( - lz4_flex::block::decompress_size_prepended(value).ok()?, - ), + Codec::Lz4 => { + 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, E> + ) -> Result> where P: ConnectionProvider, F: FnOnce(Vec) -> Fut, Fut: Future, E>>, - E: From, + 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, E> + ) -> Result> where P: ConnectionProvider, F: FnOnce(Vec) -> Fut, Fut: Future, E>>, - E: From, + 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) - .await? - .into_iter() - .map(|(key, value)| (key, (None::, value))) - .collect()) + let values = match closure(ids).await { + Ok(values) => values, + Err(error) => return Err(error), + }; + Ok::<_, E>( + values + .into_iter() + .map(|(key, value)| (key, (None::, value))) + .collect(), + ) }, ) .await @@ -274,12 +285,12 @@ impl CacheManager { case_sensitive: bool, keys: &[I], closure: F, - ) -> Result, E> + ) -> Result> where P: ConnectionProvider, F: FnOnce(Vec) -> Fut, Fut: Future, T)>, E>>, - E: From, + 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, E> + ) -> Result> where P: ConnectionProvider, F: FnOnce(Vec) -> Fut, Fut: Future, T)>, E>>, - E: From, + 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::>(); 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)?; - Ok::<_, E>( + .wrap_err("fetching Redis cache slug values")?; + 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::>(); - let mut connection = - provider.connect().await.map_err(E::from)?; + let mut connection = provider + .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 = - self.settings.encode_value(&value).map_err(E::from)?; + let encoded = self + .settings + .encode_value(&value) + .wrap_err("encoding Redis cache value")?; encoded_values.push((key, slug, value, encoded)); } - let mut connection = - provider.connect().await.map_err(E::from)?; + let mut connection = provider + .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, identity: String) { async fn wait_for_locks( waiters: Vec<(I, LockWaiter)>, deadline: Instant, -) -> Result, Error> { +) -> Result> { 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( Ok(()) => { released.push(key); } - Err(error) - if is_lock_timeout(&error) || Instant::now() >= deadline => - { + Err(_) if 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 { - matches!(error, Error::LocalCacheTimeout { .. }) -} - -fn lock_timeout_error(locks_released: usize, locks_waiting: usize) -> Error { - Error::LocalCacheTimeout { - released: locks_released, - total: locks_waiting, - } +fn lock_timeout_error( + locks_released: usize, + locks_waiting: usize, +) -> eyre::Report { + eyre!( + "timeout waiting on local Redis cache lock ({locks_released}/{locks_waiting} released)" + ) } #[derive(Serialize, Deserialize)] diff --git a/packages/xredis/src/cache/locking/local.rs b/packages/xredis/src/cache/locking/local.rs index fab5f6f822..001a64f72b 100644 --- a/packages/xredis/src/cache/locking/local.rs +++ b/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>>, @@ -76,10 +75,7 @@ pub(in crate::cache) struct LockWaiter { } impl LockWaiter { - pub(in crate::cache) async fn wait( - self, - deadline: Instant, - ) -> Result<(), Error> { + pub(in crate::cache) async fn wait(self, deadline: Instant) -> Result<()> { 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, - } -} diff --git a/packages/xredis/src/commands.rs b/packages/xredis/src/commands.rs index fdf0b058c7..744e158a89 100644 --- a/packages/xredis/src/commands.rs +++ b/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( 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( data: D, expiry: Option, 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( - connection: &mut C, - key: &str, -) -> Result, Error> +pub async fn get(connection: &mut C, key: &str) -> Result> 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( connection: &mut C, keys: &[String], -) -> Result>>, Error> +) -> Result>>> where C: ConnectionLike, { @@ -83,7 +86,7 @@ where pub async fn get_many_strings( connection: &mut C, keys: &[String], -) -> Result>, Error> +) -> Result>> where C: ConnectionLike, { @@ -93,7 +96,7 @@ where pub(super) async fn get_many_primary( connection: &mut C, keys: &[String], -) -> Result>>, Error> +) -> Result>>> where C: RoutableConnection, { @@ -103,7 +106,7 @@ where pub(super) async fn get_many_strings_primary( connection: &mut C, keys: &[String], -) -> Result>, Error> +) -> Result>> where C: RoutableConnection, { @@ -113,7 +116,7 @@ where pub(super) async fn get_many_as( connection: &mut C, keys: &[String], -) -> Result>, Error> +) -> Result>> where C: ConnectionLike, T: FromRedisValue, @@ -123,7 +126,8 @@ where let part = cmd("MGET") .arg(chunk) .query_async::>>(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( connection: &mut C, keys: &[String], -) -> Result>, Error> +) -> Result>> where C: RoutableConnection, T: FromRedisValue, @@ -143,10 +147,14 @@ where command.arg(chunk); let value = connection .route_command(command, primary_mget_routing(chunk)) - .await?; - let part = - redis::from_redis_value::>>(value.extract_error()?) - .map_err(redis::RedisError::from)?; + .await + .wrap_err("fetching multiple values from primary Redis nodes")?; + let value = value + .extract_error() + .wrap_err("extracting Redis response")?; + let part = redis::from_redis_value::>>(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( connection: &mut C, key: &str, settings: &CacheSettings, -) -> Result, Error> +) -> Result> where C: ConnectionLike, R: for<'a> serde::Deserialize<'a>, { - let value: Option> = - cmd("GET").arg(key).query_async(connection).await?; + let value: Option> = cmd("GET") + .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( connection: &mut C, keys: &[String], settings: &CacheSettings, -) -> Result>, Error> +) -> Result>> 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(connection: &mut C, key: &str) -> Result<(), Error> +pub async fn delete(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( - connection: &mut C, - keys: &[String], -) -> Result<(), Error> +pub async fn delete_many(connection: &mut C, keys: &[String]) -> Result<()> 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( - connection: &mut C, - key: &str, - value: D, -) -> Result<(), Error> +pub async fn lpush(connection: &mut C, key: &str, value: D) -> Result<()> 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( - connection: &mut C, - key: &str, -) -> Result, Error> +pub async fn incr(connection: &mut C, key: &str) -> Result> 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") } diff --git a/packages/xredis/src/config.rs b/packages/xredis/src/config.rs index 6e466b107d..d63b3b7f89 100644 --- a/packages/xredis/src/config.rs +++ b/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( - name: &'static str, - max: usize, - min: usize, - ) -> Result { + fn new(name: &'static str, max: usize, min: usize) -> Result { 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 { + ) -> Result { 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::>(); 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( - "standalone", - standalone_pool_size.0, - standalone_pool_size.1, - )?) + RedisBackendConfig::StandalonePooled( + RedisPoolSize::new( + "standalone", + standalone_pool_size.0, + standalone_pool_size.1, + ) + .wrap_err("validating standalone Redis pool size")?, + ) } (RedisTopology::Cluster, RedisConnectionType::Pooled) => { - RedisBackendConfig::ClusterPooled(RedisPoolSize::new( - "cluster", - cluster_pool_size.0, - cluster_pool_size.1, - )?) + RedisBackendConfig::ClusterPooled( + RedisPoolSize::new( + "cluster", + 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, }) diff --git a/packages/xredis/src/connection.rs b/packages/xredis/src/connection.rs index 0d66fc818d..3632290d19 100644 --- a/packages/xredis/src/connection.rs +++ b/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( - config: &RedisConfig, - ) -> Result { + pub(crate) async fn new(config: &RedisConfig) -> Result { 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 { + ) -> Result { 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 { + ) -> Result { 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( - config: &RedisConfig, - ) -> Result { + async fn cluster_multiplexed(config: &RedisConfig) -> Result { 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 connection = client.get_async_connection().await?; + let client = builder + .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( - &self, - ) -> Result { + pub(crate) async fn connect(&self) -> Result { let inner = match self { Self::StandalonePooled(pool) => { - RedisConnectionInner::StandalonePooled(pool.get().await?) - } - Self::ClusterPooled(pool) => { - RedisConnectionInner::ClusterPooled(pool.get().await?) + RedisConnectionInner::StandalonePooled( + 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( - &self, - registry: &Registry, - ) -> Result<(), prometheus::Error> { + pub(crate) fn register_metrics(&self, registry: &Registry) -> Result<()> { 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> { - let connections = try_join_all((0..min).map(|_| pool.get())).await?; +) -> Result<()> { + 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> { - let connections = try_join_all((0..min).map(|_| pool.get())).await?; +) -> Result<()> { + let connections = try_join_all((0..min).map(|_| pool.get())) + .await + .wrap_err("fetching initial clustered Redis connections")?; drop(connections); Ok(()) } diff --git a/packages/xredis/src/lib.rs b/packages/xredis/src/lib.rs index e84fce15eb..19d4afce1e 100644 --- a/packages/xredis/src/lib.rs +++ b/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>, config: RedisConfig, cache_settings: CacheSettings, - ) -> Result { + ) -> Result { 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 { + pub async fn connect(&self) -> Result { 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> { - self.backend.register_metrics(registry)?; - self.blocking.register_metrics(registry) + ) -> Result<()> { + self.backend + .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( @@ -123,11 +118,11 @@ impl RedisPool { namespace: &str, keys: &[K], closure: F, - ) -> Result, E> + ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, E>>, - E: From, + 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, E> + ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, E>>, - E: From, + 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, E> + ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, T)>, E>>, - E: From, + 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, E> + ) -> Result> where F: FnOnce(Vec) -> Fut, Fut: Future, T)>, E>>, - E: From, + 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( - &self, - ) -> impl Future> + Send { + fn connect(&self) -> impl Future> + Send { RedisPool::connect(self) } } @@ -259,7 +252,7 @@ impl RedisConnection { key: &str, data: D, expiry: Option, - ) -> Result<(), Error> + ) -> Result<()> where D: ToRedisArgs + Send + Sync + Debug, { @@ -277,7 +270,7 @@ impl RedisConnection { key: &str, data: D, expiry: Option, - ) -> Result<(), Error> + ) -> Result<()> where D: Serialize, { @@ -291,31 +284,28 @@ impl RedisConnection { .await } - pub async fn get(&mut self, key: &str) -> Result, Error> { + pub async fn get(&mut self, key: &str) -> Result> { commands::get(&mut self.inner, key).await } pub async fn get_many( &mut self, keys: &[String], - ) -> Result>>, Error> { + ) -> Result>>> { commands::get_many(&mut self.inner, keys).await } pub async fn get_many_typed( &mut self, keys: &[String], - ) -> Result>, Error> + ) -> Result>> where R: FromRedisValue, { commands::get_many_as(&mut self.inner, keys).await } - pub async fn get_deserialized( - &mut self, - key: &str, - ) -> Result, Error> + pub async fn get_deserialized(&mut self, key: &str) -> Result> where R: for<'a> serde::Deserialize<'a>, { @@ -325,7 +315,7 @@ impl RedisConnection { pub async fn get_many_deserialized( &mut self, keys: &[String], - ) -> Result>, Error> + ) -> Result>> 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(&mut self, key: &str, value: D) -> Result<(), Error> + pub async fn lpush(&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, Error> { + pub async fn incr(&mut self, key: &str) -> Result> { commands::incr(&mut self.inner, key).await } } diff --git a/packages/xredis/src/metrics.rs b/packages/xredis/src/metrics.rs index d50e4cce47..fafd5dccbe 100644 --- a/packages/xredis/src/metrics.rs +++ b/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 { + ) -> Result { 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.register(Box::new(size.clone()))?; - registry.register(Box::new(available.clone()))?; - registry.register(Box::new(waiting.clone()))?; + registry + .register(Box::new(max_size.clone())) + .wrap_err("registering Redis pool maximum size metric")?; + 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

( registry: &Registry, provider: P, -) -> Result<(), prometheus::Error> +) -> Result<()> where P: LogicalPoolStatusProvider, { @@ -132,7 +145,7 @@ where pub(super) fn register_blocking_pool_metrics

( registry: &Registry, provider: P, -) -> Result<(), prometheus::Error> +) -> Result<()> where P: LogicalPoolStatusProvider, { @@ -143,11 +156,12 @@ fn register_pool_metrics

( 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 { diff --git a/packages/xredis/src/pubsub.rs b/packages/xredis/src/pubsub.rs index b6080a0f32..c4986ce043 100644 --- a/packages/xredis/src/pubsub.rs +++ b/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( - &self, - channel: &str, - message: M, - ) -> Result<(), Error> + pub async fn publish(&self, channel: &str, message: M) -> Result<()> 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>, -) -> redis::RedisResult { - let client = redis::Client::open(seed_url)?; - let mut pubsub = client.get_async_pubsub().await?; - pubsub.subscribe(channel).await?; +) -> Result { + let client = redis::Client::open(seed_url) + .wrap_err("configuring Redis Pub/Sub client")?; + 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();