filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Redis Sentinel support as an in-memory store and cluster coordinator backend (closes #2430)

Browse Source
This commit is contained in:
Maurus Decimus
2026-06-25 10:26:23 +02:00
parent ab7d81e486
commit 93b3b70616
27 changed files with 527 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
crates/store/Cargo.toml
View File

@@ -10,7 +10,7 @@ nlp = { path = "../nlp" }
trc = { path = "../trc" }
registry = { path = "../registry" }
rocksdb = { version = "0.24", optional = true, features = ["multi-threaded-cf"] }
foundationdb = { version = "0.10", features = ["embedded-fdb-include", "fdb-7_4"], optional = true }
foundationdb = { version = "0.11", features = ["embedded-fdb-include", "fdb-7_4"], optional = true }
rusqlite = { version = "0.40", features = ["bundled"], optional = true }
rust-s3 = { version = "0.37", default-features = false, features = ["tokio-rustls-tls"], optional = true }
reqwest_s3 = { package = "reqwest", version = "0.12", default-features = false, features = ["rustls-tls-native-roots"], optional = true }
@@ -44,8 +44,8 @@ mysql_async = { version = "0.36", default-features = false, features = ["default
serde_json = { version = "1.0.64" }
regex = "1.12"
flate2 = "1.1"
redis = { version = "1.1", features = [ "tokio-comp", "tokio-rustls-comp", "tls-rustls-insecure", "tls-rustls", "cluster-async"], optional = true }
deadpool = { version = "0.12", features = ["managed"], optional = true }
redis = { version = "1.1", features = [ "tokio-comp", "tokio-rustls-comp", "tls-rustls-insecure", "tls-rustls", "cluster-async", "sentinel"], optional = true }
deadpool = { version = "0.13", features = ["managed"], optional = true }
arc-swap = "1.6.0"
bitpacking = "0.9.2"
memchr = { version = "2.7" }
@@ -71,7 +71,7 @@ s3 = ["rust-s3", "dep:reqwest_s3"]
azure = ["azure_core", "azure_storage", "azure_storage_blobs", "futures"]
# In-memory stores
redis = ["dep:redis", "deadpool", "futures"]
redis = ["dep:redis", "deadpool", "deadpool/rt_tokio_1", "futures"]
enterprise = []
test_mode = []

7
crates/store/src/backend/postgres/main.rs
View File

@@ -14,8 +14,9 @@ use crate::{
*,
};
use ::registry::schema::{enums::PostgreSqlRecyclingMethod, structs};
use deadpool::managed::Object;
use deadpool_postgres::{Config, Manager, ManagerConfig, PoolConfig, RecyclingMethod, Runtime};
use deadpool_postgres::{
Config, ManagerConfig, Object, PoolConfig, RecyclingMethod, Runtime,
};
use tokio_postgres::NoTls;
use utils::tls::rustls_client_config;
@@ -169,7 +170,7 @@ impl PostgresStore {
}
async fn create_search_tables<T: SearchableField + PsqlSearchField + 'static>(
conn: &Object<Manager>,
conn: &Object,
) -> trc::Result<()> {
let table_name = T::index().psql_table();
let mut query = format!("CREATE TABLE IF NOT EXISTS {} (", table_name);

2
crates/store/src/backend/postgres/mod.rs
View File

@@ -42,7 +42,7 @@ fn into_error(err: tokio_postgres::error::Error) -> trc::Error {
}
#[inline(always)]
fn into_pool_error(err: deadpool::managed::PoolError<tokio_postgres::Error>) -> trc::Error {
fn into_pool_error(err: deadpool_postgres::PoolError) -> trc::Error {
trc::StoreEvent::PostgresqlError.reason(err)
}

44
crates/store/src/backend/redis/lookup.rs
View File

@@ -4,11 +4,9 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use redis::AsyncCommands;
use crate::Deserialize;
use super::{RedisPool, RedisStore, into_error};
use crate::Deserialize;
use redis::AsyncCommands;
impl RedisStore {
pub async fn key_set(&self, key: &[u8], value: &[u8], expires: Option<u64>) -> trc::Result<()> {
@@ -31,6 +29,15 @@ impl RedisStore {
)
.await
}
RedisPool::Sentinel(pool) => {
self.key_set_(
pool.get().await.map_err(into_error)?.as_mut(),
key,
value,
expires,
)
.await
}
}
}
@@ -54,6 +61,15 @@ impl RedisStore {
)
.await
}
RedisPool::Sentinel(pool) => {
self.key_incr_(
pool.get().await.map_err(into_error)?.as_mut(),
key,
value,
expires,
)
.await
}
}
}
@@ -67,6 +83,10 @@ impl RedisStore {
self.key_delete_(pool.get().await.map_err(into_error)?.as_mut(), key)
.await
}
RedisPool::Sentinel(pool) => {
self.key_delete_(pool.get().await.map_err(into_error)?.as_mut(), key)
.await
}
}
}
@@ -80,6 +100,10 @@ impl RedisStore {
self.key_delete_prefix_(pool.get().await.map_err(into_error)?.as_mut(), prefix)
.await
}
RedisPool::Sentinel(pool) => {
self.key_delete_prefix_(pool.get().await.map_err(into_error)?.as_mut(), prefix)
.await
}
}
}
@@ -96,6 +120,10 @@ impl RedisStore {
self.key_get_(pool.get().await.map_err(into_error)?.as_mut(), key)
.await
}
RedisPool::Sentinel(pool) => {
self.key_get_(pool.get().await.map_err(into_error)?.as_mut(), key)
.await
}
}
}
@@ -109,6 +137,10 @@ impl RedisStore {
self.counter_get_(pool.get().await.map_err(into_error)?.as_mut(), key)
.await
}
RedisPool::Sentinel(pool) => {
self.counter_get_(pool.get().await.map_err(into_error)?.as_mut(), key)
.await
}
}
}
@@ -122,6 +154,10 @@ impl RedisStore {
self.key_exists_(pool.get().await.map_err(into_error)?.as_mut(), key)
.await
}
RedisPool::Sentinel(pool) => {
self.key_exists_(pool.get().await.map_err(into_error)?.as_mut(), key)
.await
}
}
}

64
crates/store/src/backend/redis/mod.rs
View File

@@ -10,9 +10,10 @@ use deadpool::{
managed::{Manager, Pool},
};
use redis::{
Client, ProtocolVersion,
Client, IntoConnectionInfo, ProtocolVersion,
cluster::{ClusterClient, ClusterClientBuilder},
cluster_read_routing::RandomReplicaStrategy,
sentinel::{SentinelClient, SentinelClientBuilder, SentinelServerType},
};
use registry::{
schema::{enums::RedisProtocol, structs},
@@ -38,9 +39,15 @@ pub struct RedisClusterConnectionManager {
timeout: std::time::Duration,
}
pub struct RedisSentinelConnectionManager {
pub client: tokio::sync::Mutex<SentinelClient>,
timeout: std::time::Duration,
}
pub enum RedisPool {
Single(Pool<RedisConnectionManager>),
Cluster(Pool<RedisClusterConnectionManager>),
Sentinel(Pool<RedisSentinelConnectionManager>),
}
impl RedisStore {
@@ -101,6 +108,60 @@ impl RedisStore {
)?),
})))
}
pub async fn open_sentinel(
config: structs::RedisSentinelStore,
) -> Result<InMemoryStore, String> {
let mut sentinels = Vec::with_capacity(config.urls.len());
for url in config.urls {
let info = url
.into_connection_info()
.map_err(|err| format!("Invalid Redis Sentinel URL: {err}"))?;
sentinels.push(info.addr().clone());
}
let mut builder =
SentinelClientBuilder::new(sentinels, config.service_name, SentinelServerType::Master)
.map_err(|err| format!("Failed to create Redis Sentinel client: {err:?}"))?;
if let Some(value) = config.auth_username {
builder = builder.set_client_to_redis_username(value);
}
if let Some(value) = config.auth_secret.secret().await?.map(|v| v.into_owned()) {
builder = builder.set_client_to_redis_password(value);
}
if let Some(value) = config.sentinel_username {
builder = builder.set_client_to_sentinel_username(value);
}
if let Some(value) = config
.sentinel_secret
.secret()
.await?
.map(|v| v.into_owned())
{
builder = builder.set_client_to_sentinel_password(value);
}
if matches!(config.protocol_version, RedisProtocol::Resp3) {
builder = builder.set_client_to_redis_protocol(ProtocolVersion::RESP3);
}
let client = builder
.build()
.map_err(|err| format!("Failed to open Redis Sentinel client: {err:?}"))?;
Ok(InMemoryStore::Redis(Arc::new(RedisStore {
pool: RedisPool::Sentinel(build_pool(
RedisSentinelConnectionManager {
client: tokio::sync::Mutex::new(client),
timeout: config.timeout.into_inner(),
},
config.pool_max_connections,
config.pool_timeout_create,
config.pool_timeout_wait,
config.pool_timeout_recycle,
)?),
})))
}
}
fn build_pool<M: Manager>(
@@ -130,6 +191,7 @@ impl std::fmt::Debug for RedisPool {
match self {
Self::Single(_) => f.debug_tuple("Single").finish(),
Self::Cluster(_) => f.debug_tuple("Cluster").finish(),
Self::Sentinel(_) => f.debug_tuple("Sentinel").finish(),
}
}
}

30
crates/store/src/backend/redis/pool.rs
View File

@@ -4,14 +4,16 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use super::{
RedisClusterConnectionManager, RedisConnectionManager, RedisSentinelConnectionManager,
into_error,
};
use deadpool::managed;
use redis::{
aio::{ConnectionLike, MultiplexedConnection},
cluster_async::ClusterConnection,
};
use super::{RedisClusterConnectionManager, RedisConnectionManager, into_error};
impl managed::Manager for RedisConnectionManager {
type Type = MultiplexedConnection;
type Error = trc::Error;
@@ -59,3 +61,27 @@ impl managed::Manager for RedisClusterConnectionManager {
.map_err(|err| managed::RecycleError::Backend(into_error(err)))
}
}
impl managed::Manager for RedisSentinelConnectionManager {
type Type = MultiplexedConnection;
type Error = trc::Error;
async fn create(&self) -> Result<MultiplexedConnection, trc::Error> {
let mut client = self.client.lock().await;
match tokio::time::timeout(self.timeout, client.get_async_connection()).await {
Ok(conn) => conn.map_err(into_error),
Err(_) => Err(trc::StoreEvent::RedisError.ctx(trc::Key::Details, "Connection Timeout")),
}
}
async fn recycle(
&self,
conn: &mut MultiplexedConnection,
_: &managed::Metrics,
) -> managed::RecycleResult<trc::Error> {
conn.req_packed_command(&redis::cmd("PING"))
.await
.map(|_| ())
.map_err(|err| managed::RecycleError::Backend(into_error(err)))
}
}

4
crates/store/src/build/memory.rs
View File

@@ -22,6 +22,10 @@ impl InMemoryStore {
structs::InMemoryStore::RedisCluster(redis_cluster_store) => {
crate::backend::redis::RedisStore::open_cluster(redis_cluster_store).await
}
#[cfg(feature = "redis")]
structs::InMemoryStore::RedisSentinel(redis_sentinel_store) => {
crate::backend::redis::RedisStore::open_sentinel(redis_sentinel_store).await
}
// SPDX-SnippetBegin
// SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
// SPDX-License-Identifier: LicenseRef-SEL