sugoidogo/modrinth
1
0
Fork 0
mirror of https://github.com/modrinth/code.git synced 2026-08-25 09:04:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: use postcard for redis serde (#6956)

Browse Source 
* redo error handling in xredis

* give proper types to metadata fields

* add round-trip tests

* inline loader enum metadata fields

* postcard roundtrips

* prepare

* bump redis key version

* serde-binhum

* clippy

* fix

* fix frontend checking existence of component fields rather than non-null-ness

* prepare
This commit is contained in:
aecsocket
2026-08-05 10:36:05 +00:00
committed by GitHub
parent cc53d9a6f8
commit 5148e8ec35
70 changed files with 1713 additions and 646 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/xredis/src/blocking.rs
+34 -24
View File
@@ -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<Self, RedisBackendBuildError> {
pub(super) async fn new(config: &RedisConfig) -> Result<Self> {
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<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 {
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<u8>, Vec<u8>)> =
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<Option<[Vec<u8>; 2]>, Error> {
) -> Result<Option<[Vec<u8>; 2]>> {
self.blocking.brpop(key, timeout).await
}
}
packages/xredis/src/cache.rs
+97 -65
View File
@@ -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<Output = Result<Self::Connection, Error>> + Send;
fn connect(&self) -> impl Future<Output = Result<Self::Connection>> + 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>(
&self,
value: &T,
) -> Result<Vec<u8>, Error> {
pub fn encode_value<T: Serialize>(&self, value: &T) -> Result<Vec<u8>> {
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
@@ -149,15 +144,23 @@ impl CacheSettings {
T: for<'a> Deserialize<'a>,
{
let (codec, value) = value.split_first()?;
let value = match Codec::try_from(*codec).ok()? {
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 +205,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 +223,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 +236,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 +259,16 @@ impl CacheManager {
false,
keys,
|ids| async move {
Ok(closure(ids)
.await?
.into_iter()
.map(|(key, value)| (key, (None::<String>, 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::<String>, value)))
.collect(),
)
},
)
.await
@@ -274,12 +283,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 +309,8 @@ impl CacheManager {
keys,
closure,
)
.await?
.await
.wrap_err("fetching Redis cache values by slug")?
.into_values()
.collect())
}
@@ -314,12 +324,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 +370,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 +389,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 +398,8 @@ impl CacheManager {
)
}
.instrument(info_span!("get slug ids"))
.await?
.await
.wrap_err("resolving Redis cache slugs")?
} else {
Vec::new()
};
@@ -403,8 +416,10 @@ impl CacheManager {
.map(|key| self.key_builder.entity(namespace, key))
.collect::<Vec<_>>();
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 +430,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 +440,7 @@ impl CacheManager {
}
}
Ok::<_, E>((cached_values, ids))
eyre::Ok((cached_values, ids))
}
.instrument(info_span!("get_cached_values_closure"))
};
@@ -437,7 +452,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 +525,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 +540,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 +561,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 +578,7 @@ impl CacheManager {
default_expiry,
)
.await
.map_err(E::from)?;
.wrap_err("writing Redis cache slug")?;
}
}
@@ -563,7 +587,7 @@ impl CacheManager {
return_values.insert(key, value);
}
Result::<_, E>::Ok(return_values)
Result::<_>::Ok(return_values)
}
.await
} else {
@@ -604,13 +628,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 +704,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 +720,24 @@ async fn wait_for_locks<I>(
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)]
@@ -726,6 +749,15 @@ pub struct RedisValue<T, K, S> {
}
impl<T, K, S> RedisValue<T, K, S> {
pub fn new(key: K, alias: Option<S>, iat: i64, val: T) -> Self {
Self {
key,
alias,
iat,
val,
}
}
pub fn value(&self) -> &T {
&self.val
}
packages/xredis/src/cache/locking/local.rs
Vendored
+3 -14
View File
@@ -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(
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,
}
}
packages/xredis/src/commands.rs
+60 -45
View File
@@ -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>(
connection: &mut C,
key: &str,
) -> Result<Option<String>, Error>
pub async fn get<C>(connection: &mut C, key: &str) -> Result<Option<String>>
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?;
let part =
redis::from_redis_value::<Vec<Option<T>>>(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::<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>> =
cmd("GET").arg(key).query_async(connection).await?;
let value: Option<Vec<u8>> = 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<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>(
connection: &mut C,
keys: &[String],
) -> Result<(), Error>
pub async fn delete_many<C>(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<C, D>(
connection: &mut C,
key: &str,
value: D,
) -> Result<(), Error>
pub async fn lpush<C, D>(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<C>(
connection: &mut C,
key: &str,
) -> Result<Option<u64>, Error>
pub async fn incr<C>(connection: &mut C, key: &str) -> Result<Option<u64>>
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
+30 -22
View File
@@ -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<Self, RedisConfigError> {
fn new(name: &'static str, max: usize, min: usize) -> Result<Self> {
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(
"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,
})
packages/xredis/src/connection.rs
+46 -42
View File
@@ -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<Self, RedisBackendBuildError> {
pub(crate) async fn new(config: &RedisConfig) -> Result<Self> {
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(
config: &RedisConfig,
) -> Result<Self, RedisBackendBuildError> {
async fn cluster_multiplexed(config: &RedisConfig) -> Result<Self> {
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<RedisConnection, deadpool_redis::PoolError> {
pub(crate) async fn connect(&self) -> Result<RedisConnection> {
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(())
}
packages/xredis/src/lib.rs
+41 -51
View File
@@ -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> {
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<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(
&self,
) -> impl Future<Output = Result<Self::Connection, Error>> + Send {
fn connect(&self) -> impl Future<Output = Result<Self::Connection>> + 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>(
&mut self,
key: &str,
) -> Result<Option<R>, Error>
pub async fn get_deserialized<R>(&mut self, key: &str) -> Result<Option<R>>
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
+27 -13
View File
@@ -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.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<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
+19 -12
View File
@@ -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>(
&self,
channel: &str,
message: M,
) -> Result<(), Error>
pub async fn publish<M>(&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<Vec<u8>>,
) -> redis::RedisResult<SubscriptionOutcome> {
let client = redis::Client::open(seed_url)?;
let mut pubsub = client.get_async_pubsub().await?;
pubsub.subscribe(channel).await?;
) -> Result<SubscriptionOutcome> {
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();