diff --git a/CHANGELOG.md b/CHANGELOG.md index 1bf07a42..175f3478 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ If you are upgrading from v0.16.x, replace the binary (or run `docker pull`). If ## Added - Encryption-at-rest: Support for `AES-256-GCM` and `ChaCha20-Poly1305` for S/MIME (#161). - S3: Support for `allowInvalidCerts` option to allow connecting to S3 endpoints with invalid TLS certificates. +- Redis Sentinel support as an in-memory store and cluster coordinator backend (#2430). ## Changed diff --git a/Cargo.lock b/Cargo.lock index 8aec4406..436d6481 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1858,24 +1858,23 @@ dependencies = [ [[package]] name = "deadpool" -version = "0.10.0" +version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb84100978c1c7b37f09ed3ce3e5f843af02c2a2c431bae5b19230dad2c1b490" +checksum = "0be2b1d1d6ec8d846f05e137292d0b89133caf95ef33695424c09568bdd39b1b" dependencies = [ - "async-trait", - "deadpool-runtime", + "deadpool-runtime 0.1.4", + "lazy_static", "num_cpus", "tokio", ] [[package]] name = "deadpool" -version = "0.12.3" +version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0be2b1d1d6ec8d846f05e137292d0b89133caf95ef33695424c09568bdd39b1b" +checksum = "883466cb8db62725aee5f4a6011e8a5d42912b42632df32aad57fc91127c6e04" dependencies = [ - "deadpool-runtime", - "lazy_static", + "deadpool-runtime 0.3.1", "num_cpus", "tokio", ] @@ -1903,6 +1902,15 @@ dependencies = [ "tokio", ] +[[package]] +name = "deadpool-runtime" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2657f61fb1dd8bf37a8d51093cc7cee4e77125b22f7753f49b289f831bec2bae" +dependencies = [ + "tokio", +] + [[package]] name = "decancer" version = "3.3.3" @@ -2033,10 +2041,9 @@ version = "0.16.11" dependencies = [ "ahash", "argon2", - "async-trait", "base64 0.22.1", "compact_str", - "deadpool 0.10.0", + "deadpool 0.13.0", "futures", "jsonwebtoken", "ldap3", @@ -2598,9 +2605,9 @@ dependencies = [ [[package]] name = "foundationdb" -version = "0.10.0" +version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "310c638e34e54f56daa986ed6c8dadecf4e4825245b73e577feb25c808222ae8" +checksum = "c9a0b9e89be4942ad3c6cf76789e7eaf9489fcca2b06b1449b2d9eb015424793" dependencies = [ "async-recursion", "async-trait", @@ -2609,7 +2616,7 @@ dependencies = [ "foundationdb-sys", "foundationdb-tuple", "futures", - "rand 0.9.4", + "rand 0.10.1", "serde", "serde_bytes", "serde_json", @@ -2619,18 +2626,18 @@ dependencies = [ [[package]] name = "foundationdb-gen" -version = "0.10.0" +version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b749fc82bb7576794d4679f166a5cfa0a856122e05635437f7231e5e679fa2f6" +checksum = "db58c729fca985f56d347e5f8df5c1d3135cc8211e5688eacf695e1365449594" dependencies = [ "xml-rs", ] [[package]] name = "foundationdb-macros" -version = "0.3.3" +version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4ca5370149145ec3741cd7e82832f17f893b9421ee4e484d9511c6702bd9911" +checksum = "0f51be872cad567c58cf475749b25b9bb96e9ab6063af8bbe4f965ecf40e13cb" dependencies = [ "proc-macro2", "quote", @@ -2640,9 +2647,9 @@ dependencies = [ [[package]] name = "foundationdb-sys" -version = "0.10.0" +version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0f3f231ff5b8465063b2bf8af3b975222b4fa3cc8c07b43d01fe8e0ae15c9dd" +checksum = "0bac3aa47fdd7af6ac1b2a8ee4684fe4ed3d6c663c7ff5888e1846a790e6601a" dependencies = [ "bindgen", "libc", @@ -2650,9 +2657,9 @@ dependencies = [ [[package]] name = "foundationdb-tuple" -version = "0.10.0" +version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a8cd42c159de9b90a91efc0ce350e3142099571ef498e7fa1a1dcc1d1eeccd9d" +checksum = "3c5262f7a16bddd129d31d96d09de25dee2295ce7cccaa7ed6b212f391819c64" dependencies = [ "memchr", "uuid", @@ -7787,7 +7794,7 @@ dependencies = [ "blake3", "bytes", "compact_str", - "deadpool 0.12.3", + "deadpool 0.13.0", "deadpool-postgres", "farmhash", "flate2", diff --git a/crates/coordinator/Cargo.toml b/crates/coordinator/Cargo.toml index bd43fff3..e84503ae 100644 --- a/crates/coordinator/Cargo.toml +++ b/crates/coordinator/Cargo.toml @@ -13,7 +13,7 @@ tokio = { version = "1.47", features = ["sync", "fs", "io-util"] } async-nats = { version = "0.49", default-features = false, features = ["server_2_10", "server_2_11", "aws-lc-rs"], optional = true } zenoh = { version = "1.3.4", default-features = false, features = ["auth_pubkey", "transport_multilink", "transport_compression", "transport_quic", "transport_tcp", "transport_tls", "transport_udp"], optional = true } rdkafka = { version = "0.39", features = ["cmake-build"], optional = true } -redis = { version = "1.1", features = [ "tokio-comp", "tokio-rustls-comp", "tls-rustls-insecure", "tls-rustls", "cluster-async"], optional = true } +redis = { version = "1.1", features = [ "tokio-comp", "tokio-rustls-comp", "tls-rustls-insecure", "tls-rustls", "cluster-async", "sentinel"], optional = true } [features] nats = ["async-nats", "futures"] diff --git a/crates/coordinator/src/backend/redis/pubsub.rs b/crates/coordinator/src/backend/redis/pubsub.rs index ec5e9322..547f1ecd 100644 --- a/crates/coordinator/src/backend/redis/pubsub.rs +++ b/crates/coordinator/src/backend/redis/pubsub.rs @@ -42,6 +42,14 @@ pub(crate) async fn redis_publish( .publish(topic, message) .await .map_err(into_error), + RedisPool::Sentinel(pool) => pool + .get() + .await + .map_err(into_error)? + .as_mut() + .publish(topic, message) + .await + .map_err(into_error), } } @@ -80,6 +88,23 @@ pub(crate) async fn redis_subscribe( rx, })) } + RedisPool::Sentinel(pool) => { + let client = pool + .manager() + .client + .lock() + .await + .async_get_client() + .await + .map_err(into_error)?; + + let mut pubsub = client.get_async_pubsub().await.map_err(into_error)?; + pubsub.subscribe(topic).await.map_err(into_error)?; + + Ok(PubSubStream::Redis(RedisPubSubStream { + stream: pubsub.into_on_message(), + })) + } } } diff --git a/crates/coordinator/src/bootstrap.rs b/crates/coordinator/src/bootstrap.rs index 56247f95..8cd35d67 100644 --- a/crates/coordinator/src/bootstrap.rs +++ b/crates/coordinator/src/bootstrap.rs @@ -48,6 +48,12 @@ impl Coordinator { .await .map(unwrap_redis) } + #[cfg(feature = "redis")] + structs::Coordinator::RedisSentinel(redis_sentinel_store) => { + store::backend::redis::RedisStore::open_sentinel(redis_sentinel_store) + .await + .map(unwrap_redis) + } _ => Err("Binary was not compiled with the selected coordinator backend".to_string()), }; diff --git a/crates/directory/Cargo.toml b/crates/directory/Cargo.toml index 8dca8d7e..269a2893 100644 --- a/crates/directory/Cargo.toml +++ b/crates/directory/Cargo.toml @@ -18,12 +18,11 @@ tokio-rustls = { version = "0.26", default-features = false, features = ["aws_lc rustls = { version = "0.23.5", default-features = false, features = ["std", "aws_lc_rs", "tls12"] } rustls-pki-types = { version = "1" } ldap3 = { version = "0.12", default-features = false, features = ["tls-rustls-aws-lc-rs"] } -deadpool = { version = "0.10", features = ["managed", "rt_tokio_1"] } -async-trait = "0.1.68" +deadpool = { version = "0.13", features = ["managed", "rt_tokio_1"] } ahash = { version = "0.8" } pwhash = "1" argon2 = "0.5.0" -pbkdf2 = {version = "0.12.1", features = ["simple"] } +pbkdf2 = { version = "0.12.1", features = ["simple"] } scrypt = "0.11.0" sha1 = "0.11" sha2 = "0.11" diff --git a/crates/directory/src/backend/ldap/pool.rs b/crates/directory/src/backend/ldap/pool.rs index d9df8259..a8913b79 100644 --- a/crates/directory/src/backend/ldap/pool.rs +++ b/crates/directory/src/backend/ldap/pool.rs @@ -4,13 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use async_trait::async_trait; +use super::LdapConnectionManager; use deadpool::managed; use ldap3::{Ldap, LdapConnAsync, LdapError, exop::WhoAmI}; -use super::LdapConnectionManager; - -#[async_trait] impl managed::Manager for LdapConnectionManager { type Type = Ldap; type Error = LdapError; diff --git a/crates/jmap/src/registry/set.rs b/crates/jmap/src/registry/set.rs index 7ef6ca9c..cd938206 100644 --- a/crates/jmap/src/registry/set.rs +++ b/crates/jmap/src/registry/set.rs @@ -722,7 +722,9 @@ impl RegistrySet for Server { ObjectType::AccountSettings | ObjectType::ApiKey | ObjectType::AccountPassword - | ObjectType::AppPassword => account_set(set).await.map(|set| set.into_response()), + | ObjectType::AppPassword => Box::pin(account_set(set)) + .await + .map(|set| set.into_response()), ObjectType::QueuedMessage => { queued_message_set(set).await.map(|set| set.into_response()) diff --git a/crates/registry/src/schema/enums.rs b/crates/registry/src/schema/enums.rs index 68822b80..b749e8aa 100644 --- a/crates/registry/src/schema/enums.rs +++ b/crates/registry/src/schema/enums.rs @@ -270,6 +270,7 @@ pub enum CoordinatorType { Zenoh = 4, Redis = 5, RedisCluster = 6, + RedisSentinel = 7, } #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)] @@ -848,6 +849,7 @@ pub enum InMemoryStoreBaseType { #[default] Redis = 0, RedisCluster = 1, + RedisSentinel = 2, } #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)] @@ -858,6 +860,7 @@ pub enum InMemoryStoreType { Sharded = 1, Redis = 2, RedisCluster = 3, + RedisSentinel = 4, } #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)] @@ -1265,6 +1268,7 @@ pub enum LookupStoreType { Sharded = 3, Redis = 4, RedisCluster = 5, + RedisSentinel = 6, } #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)] diff --git a/crates/registry/src/schema/enums_impl.rs b/crates/registry/src/schema/enums_impl.rs index b33bb5a0..93d045ac 100644 --- a/crates/registry/src/schema/enums_impl.rs +++ b/crates/registry/src/schema/enums_impl.rs @@ -1402,6 +1402,7 @@ impl EnumImpl for CoordinatorType { b"Zenoh" => CoordinatorType::Zenoh, b"Redis" => CoordinatorType::Redis, b"RedisCluster" => CoordinatorType::RedisCluster, + b"RedisSentinel" => CoordinatorType::RedisSentinel, } } @@ -1414,6 +1415,7 @@ impl EnumImpl for CoordinatorType { CoordinatorType::Zenoh => "Zenoh", CoordinatorType::Redis => "Redis", CoordinatorType::RedisCluster => "RedisCluster", + CoordinatorType::RedisSentinel => "RedisSentinel", } } @@ -1430,11 +1432,12 @@ impl EnumImpl for CoordinatorType { 4 => Some(CoordinatorType::Zenoh), 5 => Some(CoordinatorType::Redis), 6 => Some(CoordinatorType::RedisCluster), + 7 => Some(CoordinatorType::RedisSentinel), _ => None, } } - const COUNT: usize = 7; + const COUNT: usize = 8; } impl serde::Serialize for CoordinatorType { @@ -4036,6 +4039,7 @@ impl EnumImpl for InMemoryStoreBaseType { value.as_bytes(), b"Redis" => InMemoryStoreBaseType::Redis, b"RedisCluster" => InMemoryStoreBaseType::RedisCluster, + b"RedisSentinel" => InMemoryStoreBaseType::RedisSentinel, } } @@ -4043,6 +4047,7 @@ impl EnumImpl for InMemoryStoreBaseType { match self { InMemoryStoreBaseType::Redis => "Redis", InMemoryStoreBaseType::RedisCluster => "RedisCluster", + InMemoryStoreBaseType::RedisSentinel => "RedisSentinel", } } @@ -4054,11 +4059,12 @@ impl EnumImpl for InMemoryStoreBaseType { match id { 0 => Some(InMemoryStoreBaseType::Redis), 1 => Some(InMemoryStoreBaseType::RedisCluster), + 2 => Some(InMemoryStoreBaseType::RedisSentinel), _ => None, } } - const COUNT: usize = 2; + const COUNT: usize = 3; } impl serde::Serialize for InMemoryStoreBaseType { @@ -4088,6 +4094,7 @@ impl EnumImpl for InMemoryStoreType { b"Sharded" => InMemoryStoreType::Sharded, b"Redis" => InMemoryStoreType::Redis, b"RedisCluster" => InMemoryStoreType::RedisCluster, + b"RedisSentinel" => InMemoryStoreType::RedisSentinel, } } @@ -4097,6 +4104,7 @@ impl EnumImpl for InMemoryStoreType { InMemoryStoreType::Sharded => "Sharded", InMemoryStoreType::Redis => "Redis", InMemoryStoreType::RedisCluster => "RedisCluster", + InMemoryStoreType::RedisSentinel => "RedisSentinel", } } @@ -4110,11 +4118,12 @@ impl EnumImpl for InMemoryStoreType { 1 => Some(InMemoryStoreType::Sharded), 2 => Some(InMemoryStoreType::Redis), 3 => Some(InMemoryStoreType::RedisCluster), + 4 => Some(InMemoryStoreType::RedisSentinel), _ => None, } } - const COUNT: usize = 4; + const COUNT: usize = 5; } impl serde::Serialize for InMemoryStoreType { @@ -5489,6 +5498,7 @@ impl EnumImpl for LookupStoreType { b"Sharded" => LookupStoreType::Sharded, b"Redis" => LookupStoreType::Redis, b"RedisCluster" => LookupStoreType::RedisCluster, + b"RedisSentinel" => LookupStoreType::RedisSentinel, } } @@ -5500,6 +5510,7 @@ impl EnumImpl for LookupStoreType { LookupStoreType::Sharded => "Sharded", LookupStoreType::Redis => "Redis", LookupStoreType::RedisCluster => "RedisCluster", + LookupStoreType::RedisSentinel => "RedisSentinel", } } @@ -5515,11 +5526,12 @@ impl EnumImpl for LookupStoreType { 3 => Some(LookupStoreType::Sharded), 4 => Some(LookupStoreType::Redis), 5 => Some(LookupStoreType::RedisCluster), + 6 => Some(LookupStoreType::RedisSentinel), _ => None, } } - const COUNT: usize = 6; + const COUNT: usize = 7; } impl serde::Serialize for LookupStoreType { diff --git a/crates/registry/src/schema/properties.rs b/crates/registry/src/schema/properties.rs index b7254e72..5fc70e72 100644 --- a/crates/registry/src/schema/properties.rs +++ b/crates/registry/src/schema/properties.rs @@ -1023,10 +1023,13 @@ pub enum Property { SelectorTemplate = 226, SendFrequency = 230, SendingMtaIp = 833, + SentinelSecret = 915, + SentinelUsername = 914, Separator = 97, ServerHostname = 121, Servers = 308, ServiceAccountJson = 316, + ServiceName = 913, Services = 794, SessionToken = 329, SetMaxObjects = 440, diff --git a/crates/registry/src/schema/properties_impl.rs b/crates/registry/src/schema/properties_impl.rs index a6f27ba6..8ffc914a 100644 --- a/crates/registry/src/schema/properties_impl.rs +++ b/crates/registry/src/schema/properties_impl.rs @@ -1176,10 +1176,13 @@ impl EnumImpl for Property { b"selectorTemplate" => Property::SelectorTemplate, b"sendFrequency" => Property::SendFrequency, b"sendingMtaIp" => Property::SendingMtaIp, + b"sentinelSecret" => Property::SentinelSecret, + b"sentinelUsername" => Property::SentinelUsername, b"separator" => Property::Separator, b"serverHostname" => Property::ServerHostname, b"servers" => Property::Servers, b"serviceAccountJson" => Property::ServiceAccountJson, + b"serviceName" => Property::ServiceName, b"services" => Property::Services, b"sessionToken" => Property::SessionToken, b"setMaxObjects" => Property::SetMaxObjects, @@ -2094,10 +2097,13 @@ impl EnumImpl for Property { Property::SelectorTemplate => "selectorTemplate", Property::SendFrequency => "sendFrequency", Property::SendingMtaIp => "sendingMtaIp", + Property::SentinelSecret => "sentinelSecret", + Property::SentinelUsername => "sentinelUsername", Property::Separator => "separator", Property::ServerHostname => "serverHostname", Property::Servers => "servers", Property::ServiceAccountJson => "serviceAccountJson", + Property::ServiceName => "serviceName", Property::Services => "services", Property::SessionToken => "sessionToken", Property::SetMaxObjects => "setMaxObjects", @@ -3016,10 +3022,13 @@ impl EnumImpl for Property { 226 => Some(Property::SelectorTemplate), 230 => Some(Property::SendFrequency), 833 => Some(Property::SendingMtaIp), + 915 => Some(Property::SentinelSecret), + 914 => Some(Property::SentinelUsername), 97 => Some(Property::Separator), 121 => Some(Property::ServerHostname), 308 => Some(Property::Servers), 316 => Some(Property::ServiceAccountJson), + 913 => Some(Property::ServiceName), 794 => Some(Property::Services), 329 => Some(Property::SessionToken), 440 => Some(Property::SetMaxObjects), @@ -3164,7 +3173,7 @@ impl EnumImpl for Property { } } - const COUNT: usize = 913; + const COUNT: usize = 916; } impl serde::Serialize for Property { diff --git a/crates/registry/src/schema/structs.rs b/crates/registry/src/schema/structs.rs index b0ef5be9..7060358e 100644 --- a/crates/registry/src/schema/structs.rs +++ b/crates/registry/src/schema/structs.rs @@ -788,6 +788,7 @@ pub enum Coordinator { Zenoh(ZenohCoordinator), Redis(RedisStore), RedisCluster(RedisClusterStore), + RedisSentinel(RedisSentinelStore), } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -3075,6 +3076,7 @@ pub enum InMemoryStore { Sharded(ShardedInMemoryStore), Redis(RedisStore), RedisCluster(RedisClusterStore), + RedisSentinel(RedisSentinelStore), } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -3082,6 +3084,7 @@ pub enum InMemoryStore { pub enum InMemoryStoreBase { Redis(RedisStore), RedisCluster(RedisClusterStore), + RedisSentinel(RedisSentinelStore), } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -3258,6 +3261,7 @@ pub enum LookupStore { Sharded(ShardedInMemoryStore), Redis(RedisStore), RedisCluster(RedisClusterStore), + RedisSentinel(RedisSentinelStore), } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -4326,6 +4330,35 @@ pub struct RedisClusterStore { pub pool_timeout_recycle: Option, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(default)] +pub struct RedisSentinelStore { + #[serde(rename = "urls")] + pub urls: Map, + #[serde(rename = "serviceName")] + pub service_name: String, + #[serde(rename = "timeout")] + pub timeout: Duration, + #[serde(rename = "authUsername")] + pub auth_username: Option, + #[serde(rename = "authSecret")] + pub auth_secret: SecretKeyOptional, + #[serde(rename = "sentinelUsername")] + pub sentinel_username: Option, + #[serde(rename = "sentinelSecret")] + pub sentinel_secret: SecretKeyOptional, + #[serde(rename = "protocolVersion")] + pub protocol_version: RedisProtocol, + #[serde(rename = "poolMaxConnections")] + pub pool_max_connections: u64, + #[serde(rename = "poolTimeoutCreate")] + pub pool_timeout_create: Option, + #[serde(rename = "poolTimeoutWait")] + pub pool_timeout_wait: Option, + #[serde(rename = "poolTimeoutRecycle")] + pub pool_timeout_recycle: Option, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(default)] pub struct RedisStore { diff --git a/crates/registry/src/schema/structs_impl.rs b/crates/registry/src/schema/structs_impl.rs index e0c718d0..cc947584 100644 --- a/crates/registry/src/schema/structs_impl.rs +++ b/crates/registry/src/schema/structs_impl.rs @@ -5723,6 +5723,7 @@ impl ObjectImpl for Coordinator { Coordinator::Zenoh(inner) => inner.validate(errors), Coordinator::Redis(inner) => inner.validate(errors), Coordinator::RedisCluster(inner) => inner.validate(errors), + Coordinator::RedisSentinel(inner) => inner.validate(errors), } } @@ -5764,6 +5765,10 @@ impl Pickle for Coordinator { 6u16.pickle(out); inner.pickle(out); } + Coordinator::RedisSentinel(inner) => { + 7u16.pickle(out); + inner.pickle(out); + } } } @@ -5776,6 +5781,7 @@ impl Pickle for Coordinator { 4 => Pickle::unpickle(stream).map(Coordinator::Zenoh), 5 => Pickle::unpickle(stream).map(Coordinator::Redis), 6 => Pickle::unpickle(stream).map(Coordinator::RedisCluster), + 7 => Pickle::unpickle(stream).map(Coordinator::RedisSentinel), _ => None, } } @@ -5829,6 +5835,13 @@ impl IntoValue for Coordinator { .insert_unchecked(Property::Type, JmapValue::Str("RedisCluster".into())); obj } + Coordinator::RedisSentinel(obj) => { + let mut obj = obj.into_value(); + obj.as_object_mut() + .unwrap() + .insert_unchecked(Property::Type, JmapValue::Str("RedisSentinel".into())); + obj + } } } } @@ -5850,6 +5863,9 @@ impl RegistryJsonPatch for Coordinator { CoordinatorType::RedisCluster => { *self = Coordinator::RedisCluster(Default::default()) } + CoordinatorType::RedisSentinel => { + *self = Coordinator::RedisSentinel(Default::default()) + } } } match self { @@ -5860,6 +5876,7 @@ impl RegistryJsonPatch for Coordinator { Coordinator::Zenoh(inner) => inner.patch(pointer, value), Coordinator::Redis(inner) => inner.patch(pointer, value), Coordinator::RedisCluster(inner) => inner.patch(pointer, value), + Coordinator::RedisSentinel(inner) => inner.patch(pointer, value), } } } @@ -5874,6 +5891,7 @@ impl Coordinator { Coordinator::Zenoh(_) => CoordinatorType::Zenoh, Coordinator::Redis(_) => CoordinatorType::Redis, Coordinator::RedisCluster(_) => CoordinatorType::RedisCluster, + Coordinator::RedisSentinel(_) => CoordinatorType::RedisSentinel, } } } @@ -22312,6 +22330,7 @@ impl ObjectImpl for InMemoryStore { InMemoryStore::Sharded(inner) => inner.validate(errors), InMemoryStore::Redis(inner) => inner.validate(errors), InMemoryStore::RedisCluster(inner) => inner.validate(errors), + InMemoryStore::RedisSentinel(inner) => inner.validate(errors), } } @@ -22342,6 +22361,10 @@ impl Pickle for InMemoryStore { 3u16.pickle(out); inner.pickle(out); } + InMemoryStore::RedisSentinel(inner) => { + 4u16.pickle(out); + inner.pickle(out); + } } } @@ -22351,6 +22374,7 @@ impl Pickle for InMemoryStore { 1 => Pickle::unpickle(stream).map(InMemoryStore::Sharded), 2 => Pickle::unpickle(stream).map(InMemoryStore::Redis), 3 => Pickle::unpickle(stream).map(InMemoryStore::RedisCluster), + 4 => Pickle::unpickle(stream).map(InMemoryStore::RedisSentinel), _ => None, } } @@ -22385,6 +22409,13 @@ impl IntoValue for InMemoryStore { .insert_unchecked(Property::Type, JmapValue::Str("RedisCluster".into())); obj } + InMemoryStore::RedisSentinel(obj) => { + let mut obj = obj.into_value(); + obj.as_object_mut() + .unwrap() + .insert_unchecked(Property::Type, JmapValue::Str("RedisSentinel".into())); + obj + } } } } @@ -22403,6 +22434,9 @@ impl RegistryJsonPatch for InMemoryStore { InMemoryStoreType::RedisCluster => { *self = InMemoryStore::RedisCluster(Default::default()) } + InMemoryStoreType::RedisSentinel => { + *self = InMemoryStore::RedisSentinel(Default::default()) + } } } match self { @@ -22410,6 +22444,7 @@ impl RegistryJsonPatch for InMemoryStore { InMemoryStore::Sharded(inner) => inner.patch(pointer, value), InMemoryStore::Redis(inner) => inner.patch(pointer, value), InMemoryStore::RedisCluster(inner) => inner.patch(pointer, value), + InMemoryStore::RedisSentinel(inner) => inner.patch(pointer, value), } } } @@ -22421,6 +22456,7 @@ impl InMemoryStore { InMemoryStore::Sharded(_) => InMemoryStoreType::Sharded, InMemoryStore::Redis(_) => InMemoryStoreType::Redis, InMemoryStore::RedisCluster(_) => InMemoryStoreType::RedisCluster, + InMemoryStore::RedisSentinel(_) => InMemoryStoreType::RedisSentinel, } } } @@ -22430,6 +22466,7 @@ impl InMemoryStoreBase { match self { InMemoryStoreBase::Redis(inner) => inner.validate(errors), InMemoryStoreBase::RedisCluster(inner) => inner.validate(errors), + InMemoryStoreBase::RedisSentinel(inner) => inner.validate(errors), } } } @@ -22451,6 +22488,10 @@ impl Pickle for InMemoryStoreBase { 1u16.pickle(out); inner.pickle(out); } + InMemoryStoreBase::RedisSentinel(inner) => { + 2u16.pickle(out); + inner.pickle(out); + } } } @@ -22458,6 +22499,7 @@ impl Pickle for InMemoryStoreBase { match u16::unpickle(stream)? { 0 => Pickle::unpickle(stream).map(InMemoryStoreBase::Redis), 1 => Pickle::unpickle(stream).map(InMemoryStoreBase::RedisCluster), + 2 => Pickle::unpickle(stream).map(InMemoryStoreBase::RedisSentinel), _ => None, } } @@ -22480,6 +22522,13 @@ impl IntoValue for InMemoryStoreBase { .insert_unchecked(Property::Type, JmapValue::Str("RedisCluster".into())); obj } + InMemoryStoreBase::RedisSentinel(obj) => { + let mut obj = obj.into_value(); + obj.as_object_mut() + .unwrap() + .insert_unchecked(Property::Type, JmapValue::Str("RedisSentinel".into())); + obj + } } } } @@ -22498,11 +22547,15 @@ impl RegistryJsonPatch for InMemoryStoreBase { InMemoryStoreBaseType::RedisCluster => { *self = InMemoryStoreBase::RedisCluster(Default::default()) } + InMemoryStoreBaseType::RedisSentinel => { + *self = InMemoryStoreBase::RedisSentinel(Default::default()) + } } } match self { InMemoryStoreBase::Redis(inner) => inner.patch(pointer, value), InMemoryStoreBase::RedisCluster(inner) => inner.patch(pointer, value), + InMemoryStoreBase::RedisSentinel(inner) => inner.patch(pointer, value), } } } @@ -22512,6 +22565,7 @@ impl InMemoryStoreBase { match self { InMemoryStoreBase::Redis(_) => InMemoryStoreBaseType::Redis, InMemoryStoreBase::RedisCluster(_) => InMemoryStoreBaseType::RedisCluster, + InMemoryStoreBase::RedisSentinel(_) => InMemoryStoreBaseType::RedisSentinel, } } } @@ -23553,6 +23607,7 @@ impl LookupStore { LookupStore::Sharded(inner) => inner.validate(errors), LookupStore::Redis(inner) => inner.validate(errors), LookupStore::RedisCluster(inner) => inner.validate(errors), + LookupStore::RedisSentinel(inner) => inner.validate(errors), } } } @@ -23590,6 +23645,10 @@ impl Pickle for LookupStore { 5u16.pickle(out); inner.pickle(out); } + LookupStore::RedisSentinel(inner) => { + 6u16.pickle(out); + inner.pickle(out); + } } } @@ -23601,6 +23660,7 @@ impl Pickle for LookupStore { 3 => Pickle::unpickle(stream).map(LookupStore::Sharded), 4 => Pickle::unpickle(stream).map(LookupStore::Redis), 5 => Pickle::unpickle(stream).map(LookupStore::RedisCluster), + 6 => Pickle::unpickle(stream).map(LookupStore::RedisSentinel), _ => None, } } @@ -23651,6 +23711,13 @@ impl IntoValue for LookupStore { .insert_unchecked(Property::Type, JmapValue::Str("RedisCluster".into())); obj } + LookupStore::RedisSentinel(obj) => { + let mut obj = obj.into_value(); + obj.as_object_mut() + .unwrap() + .insert_unchecked(Property::Type, JmapValue::Str("RedisSentinel".into())); + obj + } } } } @@ -23671,6 +23738,9 @@ impl RegistryJsonPatch for LookupStore { LookupStoreType::RedisCluster => { *self = LookupStore::RedisCluster(Default::default()) } + LookupStoreType::RedisSentinel => { + *self = LookupStore::RedisSentinel(Default::default()) + } } } match self { @@ -23680,6 +23750,7 @@ impl RegistryJsonPatch for LookupStore { LookupStore::Sharded(inner) => inner.patch(pointer, value), LookupStore::Redis(inner) => inner.patch(pointer, value), LookupStore::RedisCluster(inner) => inner.patch(pointer, value), + LookupStore::RedisSentinel(inner) => inner.patch(pointer, value), } } } @@ -23693,6 +23764,7 @@ impl LookupStore { LookupStore::Sharded(_) => LookupStoreType::Sharded, LookupStore::Redis(_) => LookupStoreType::Redis, LookupStore::RedisCluster(_) => LookupStoreType::RedisCluster, + LookupStore::RedisSentinel(_) => LookupStoreType::RedisSentinel, } } } @@ -32024,6 +32096,173 @@ impl RegistryJsonPropertyPatch for RedisClusterStore { } } +impl RedisSentinelStore { + fn validate(&self, errors: &mut Vec) -> bool { + let neb = errors.len(); + let value = &self.urls; + for value in value.iter() { + if value.is_empty() { + errors.push(ValidationError::required(Property::Urls)); + } + } + let value = &self.service_name; + if value.is_empty() { + errors.push(ValidationError::required(Property::ServiceName)); + } + if let Some(value) = &self.auth_username { + if value.is_empty() { + errors.push(ValidationError::required(Property::AuthUsername)); + } + } + let value = &self.auth_secret; + value.validate(errors); + if let Some(value) = &self.sentinel_username { + if value.is_empty() { + errors.push(ValidationError::required(Property::SentinelUsername)); + } + } + let value = &self.sentinel_secret; + value.validate(errors); + let value = &self.pool_max_connections; + if *value > 8192 { + errors.push(ValidationError::max_value( + Property::PoolMaxConnections, + 8192, + )); + } + if *value < 1 { + errors.push(ValidationError::min_value(Property::PoolMaxConnections, 1)); + } + errors.len() == neb + } +} + +impl Pickle for RedisSentinelStore { + fn pickle(&self, out: &mut Vec) { + self.urls.pickle(out); + self.service_name.pickle(out); + self.timeout.pickle(out); + self.auth_username.pickle(out); + self.auth_secret.pickle(out); + self.sentinel_username.pickle(out); + self.sentinel_secret.pickle(out); + self.protocol_version.pickle(out); + self.pool_max_connections.pickle(out); + self.pool_timeout_create.pickle(out); + self.pool_timeout_wait.pickle(out); + self.pool_timeout_recycle.pickle(out); + } + + fn unpickle(stream: &mut crate::pickle::PickledStream<'_>) -> Option { + let mut this = Self::default(); + this.urls = Pickle::unpickle(stream)?; + this.service_name = Pickle::unpickle(stream)?; + this.timeout = Pickle::unpickle(stream)?; + this.auth_username = Pickle::unpickle(stream)?; + this.auth_secret = Pickle::unpickle(stream)?; + this.sentinel_username = Pickle::unpickle(stream)?; + this.sentinel_secret = Pickle::unpickle(stream)?; + this.protocol_version = Pickle::unpickle(stream)?; + this.pool_max_connections = Pickle::unpickle(stream)?; + this.pool_timeout_create = Pickle::unpickle(stream)?; + this.pool_timeout_wait = Pickle::unpickle(stream)?; + this.pool_timeout_recycle = Pickle::unpickle(stream)?; + Some(this) + } +} + +impl Default for RedisSentinelStore { + fn default() -> Self { + Self { + urls: Map::new(vec!["redis://127.0.0.1:26379".to_string()]), + service_name: "mymaster".to_string(), + timeout: Duration::from_millis(10000), + auth_username: Some("stalwart".to_string()), + auth_secret: Default::default(), + sentinel_username: Default::default(), + sentinel_secret: Default::default(), + protocol_version: RedisProtocol::Resp2, + pool_max_connections: 10u64, + pool_timeout_create: Some(Duration::from_millis(30000)), + pool_timeout_wait: Some(Duration::from_millis(30000)), + pool_timeout_recycle: Some(Duration::from_millis(30000)), + } + } +} + +impl IntoValue for RedisSentinelStore { + fn into_value(self) -> JmapValue<'static> { + let mut map = jmap_tools::Map::with_capacity(14); + map.insert_unchecked(Property::Urls, self.urls.into_value()); + map.insert_unchecked(Property::ServiceName, self.service_name.into_value()); + map.insert_unchecked(Property::Timeout, self.timeout.into_value()); + map.insert_unchecked(Property::AuthUsername, self.auth_username.into_value()); + map.insert_unchecked(Property::AuthSecret, self.auth_secret.into_value()); + map.insert_unchecked( + Property::SentinelUsername, + self.sentinel_username.into_value(), + ); + map.insert_unchecked(Property::SentinelSecret, self.sentinel_secret.into_value()); + map.insert_unchecked( + Property::ProtocolVersion, + self.protocol_version.into_value(), + ); + map.insert_unchecked( + Property::PoolMaxConnections, + self.pool_max_connections.into_value(), + ); + map.insert_unchecked( + Property::PoolTimeoutCreate, + self.pool_timeout_create.into_value(), + ); + map.insert_unchecked( + Property::PoolTimeoutWait, + self.pool_timeout_wait.into_value(), + ); + map.insert_unchecked( + Property::PoolTimeoutRecycle, + self.pool_timeout_recycle.into_value(), + ); + JmapValue::Object(map) + } +} + +impl RegistryJsonPropertyPatch for RedisSentinelStore { + fn patch_property<'x>( + &mut self, + mut pointer: JsonPointerPatch<'_>, + value: JmapValue<'x>, + ) -> PatchResult<'x> { + match pointer.next_property() { + Some(Property::Urls) => self + .urls + .patch(pointer.with_validators(&[StringValidator::Trim]), value), + Some(Property::ServiceName) => self + .service_name + .patch(pointer.with_validators(&[StringValidator::Trim]), value), + Some(Property::Timeout) => self.timeout.patch(pointer, value), + Some(Property::AuthUsername) => self + .auth_username + .patch(pointer.with_validators(&[StringValidator::Trim]), value), + Some(Property::AuthSecret) => self.auth_secret.patch(pointer, value), + Some(Property::SentinelUsername) => self + .sentinel_username + .patch(pointer.with_validators(&[StringValidator::Trim]), value), + Some(Property::SentinelSecret) => self.sentinel_secret.patch(pointer, value), + Some(Property::ProtocolVersion) => self.protocol_version.patch(pointer, value), + Some(Property::PoolMaxConnections) => self.pool_max_connections.patch(pointer, value), + Some(Property::PoolTimeoutCreate) => self.pool_timeout_create.patch(pointer, value), + Some(Property::PoolTimeoutWait) => self.pool_timeout_wait.patch(pointer, value), + Some(Property::PoolTimeoutRecycle) => self.pool_timeout_recycle.patch(pointer, value), + Some(Property::Type) => Ok(MaybeUnpatched::Unpatched { + property: Property::Type, + value, + }), + _ => Err(PatchError::new(pointer, "Invalid property")), + } + } +} + impl RedisStore { fn validate(&self, errors: &mut Vec) -> bool { let neb = errors.len(); diff --git a/crates/services/src/task_manager/acme.rs b/crates/services/src/task_manager/acme.rs index 28ee0532..37a86171 100644 --- a/crates/services/src/task_manager/acme.rs +++ b/crates/services/src/task_manager/acme.rs @@ -43,7 +43,7 @@ const MAX_RETRIES: u32 = 5; async fn acme_management(server: &Server, task: &TaskDomainManagement) -> trc::Result { let mut last_temporary_error = Ok(TaskResult::temporary("")); for retry in 0..MAX_RETRIES { - last_temporary_error = match server.acme_renew(task.domain_id).await { + last_temporary_error = match Box::pin(server.acme_renew(task.domain_id)).await { Ok(tasks) => return Ok(TaskResult::Success(tasks)), Err(err) => match err { AcmeError::Crypto(_) diff --git a/crates/smtp/src/outbound/delivery.rs b/crates/smtp/src/outbound/delivery.rs index 42bbef42..ffbb4628 100644 --- a/crates/smtp/src/outbound/delivery.rs +++ b/crates/smtp/src/outbound/delivery.rs @@ -85,7 +85,7 @@ impl QueuedMessage { // Attempt delivery let start_time = Instant::now(); - let queue_event = self.deliver_task(server.clone(), message).await; + let queue_event = Box::pin(self.deliver_task(server.clone(), message)).await; trc::event!( Delivery(DeliveryEvent::AttemptEnd), diff --git a/crates/smtp/src/scripts/exec.rs b/crates/smtp/src/scripts/exec.rs index cc5ac3f9..f3fb9d24 100644 --- a/crates/smtp/src/scripts/exec.rs +++ b/crates/smtp/src/scripts/exec.rs @@ -146,15 +146,16 @@ impl Session { script: Arc, params: ScriptParameters<'_>, ) -> ScriptResult { - self.server - .run_script( + Box::pin( + self.server.run_script( script_id, script, params .with_session_id(self.data.session_id) .with_envelope(&self.server, self, self.data.session_id) .await, - ) - .await + ), + ) + .await } } diff --git a/crates/store/Cargo.toml b/crates/store/Cargo.toml index 5e2bbbd3..56ee9520 100644 --- a/crates/store/Cargo.toml +++ b/crates/store/Cargo.toml @@ -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 = [] diff --git a/crates/store/src/backend/postgres/main.rs b/crates/store/src/backend/postgres/main.rs index e37eab92..722988a4 100644 --- a/crates/store/src/backend/postgres/main.rs +++ b/crates/store/src/backend/postgres/main.rs @@ -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( - conn: &Object, + conn: &Object, ) -> trc::Result<()> { let table_name = T::index().psql_table(); let mut query = format!("CREATE TABLE IF NOT EXISTS {} (", table_name); diff --git a/crates/store/src/backend/postgres/mod.rs b/crates/store/src/backend/postgres/mod.rs index 5152aeb0..10572032 100644 --- a/crates/store/src/backend/postgres/mod.rs +++ b/crates/store/src/backend/postgres/mod.rs @@ -42,7 +42,7 @@ fn into_error(err: tokio_postgres::error::Error) -> trc::Error { } #[inline(always)] -fn into_pool_error(err: deadpool::managed::PoolError) -> trc::Error { +fn into_pool_error(err: deadpool_postgres::PoolError) -> trc::Error { trc::StoreEvent::PostgresqlError.reason(err) } diff --git a/crates/store/src/backend/redis/lookup.rs b/crates/store/src/backend/redis/lookup.rs index 07817bb6..079c3b9e 100644 --- a/crates/store/src/backend/redis/lookup.rs +++ b/crates/store/src/backend/redis/lookup.rs @@ -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) -> 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 + } } } diff --git a/crates/store/src/backend/redis/mod.rs b/crates/store/src/backend/redis/mod.rs index 087b5025..be2f7414 100644 --- a/crates/store/src/backend/redis/mod.rs +++ b/crates/store/src/backend/redis/mod.rs @@ -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, + timeout: std::time::Duration, +} + pub enum RedisPool { Single(Pool), Cluster(Pool), + Sentinel(Pool), } impl RedisStore { @@ -101,6 +108,60 @@ impl RedisStore { )?), }))) } + + pub async fn open_sentinel( + config: structs::RedisSentinelStore, + ) -> Result { + 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( @@ -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(), } } } diff --git a/crates/store/src/backend/redis/pool.rs b/crates/store/src/backend/redis/pool.rs index bcbe07a0..341b630e 100644 --- a/crates/store/src/backend/redis/pool.rs +++ b/crates/store/src/backend/redis/pool.rs @@ -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 { + 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 { + conn.req_packed_command(&redis::cmd("PING")) + .await + .map(|_| ()) + .map_err(|err| managed::RecycleError::Backend(into_error(err))) + } +} diff --git a/crates/store/src/build/memory.rs b/crates/store/src/build/memory.rs index d3eb4552..59705b0b 100644 --- a/crates/store/src/build/memory.rs +++ b/crates/store/src/build/memory.rs @@ -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 // SPDX-License-Identifier: LicenseRef-SEL diff --git a/resources/schema/schema.json.gz b/resources/schema/schema.json.gz index d2bcb825..a8181418 100644 Binary files a/resources/schema/schema.json.gz and b/resources/schema/schema.json.gz differ diff --git a/resources/schema/schema.json.sha256 b/resources/schema/schema.json.sha256 index efdf76f6..0c7a0eb8 100644 --- a/resources/schema/schema.json.sha256 +++ b/resources/schema/schema.json.sha256 @@ -1 +1 @@ -h8p5wN6778ClyWk8ZeLEzHFmF2ujSAjRDQbktY0FJj8 \ No newline at end of file +nvdIm9lmWr6eYd0LhUUTOCKQrZ6EP2w7aPmfNXw7qzI \ No newline at end of file diff --git a/tests/Cargo.toml b/tests/Cargo.toml index c94575fe..61dd4100 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -5,8 +5,8 @@ edition = "2024" [features] #default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "nats", "azure", "foundationdb"] -#default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "foundationdb"] -default = ["rocks", "sqlite"] +default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "foundationdb"] +#default = ["rocks", "sqlite"] sqlite = ["store/sqlite", "directory/sqlite"] foundationdb = ["store/foundation", "common/foundation"] postgres = ["store/postgres", "directory/postgres"]