diff --git a/Cargo.lock b/Cargo.lock index 8b656d55..201a47a7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -369,6 +369,42 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "async-nats" +version = "0.40.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e23419d455dc57d3ae60a2f4278cf561fc74fe866e548e14d2b0ad3e1b8ca0b2" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures", + "memchr", + "nkeys", + "nuid", + "once_cell", + "pin-project", + "portable-atomic", + "rand 0.8.5", + "regex", + "ring 0.17.14", + "rustls-native-certs 0.7.3", + "rustls-pemfile 2.2.0", + "rustls-webpki 0.102.8", + "serde", + "serde_json", + "serde_nanos", + "serde_repr", + "thiserror 1.0.69", + "time", + "tokio", + "tokio-rustls 0.26.2", + "tokio-util", + "tokio-websockets", + "tracing", + "tryhard", + "url", +] + [[package]] name = "async-recursion" version = "1.1.1" @@ -966,6 +1002,9 @@ name = "bytes" version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a" +dependencies = [ + "serde", +] [[package]] name = "bzip2" @@ -2114,6 +2153,7 @@ dependencies = [ "rand_core 0.6.4", "serde", "sha2 0.10.9", + "signature", "subtle", "zeroize", ] @@ -4571,6 +4611,21 @@ dependencies = [ "libc", ] +[[package]] +name = "nkeys" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f49e787f4c61cbd0f9320b31cc26e58719f6aa5068e34697dd3aea361412fe3" +dependencies = [ + "data-encoding", + "ed25519", + "ed25519-dalek", + "getrandom 0.2.16", + "log", + "rand 0.8.5", + "signatory", +] + [[package]] name = "nlp" version = "0.11.8" @@ -4621,6 +4676,15 @@ dependencies = [ "winapi", ] +[[package]] +name = "nuid" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc895af95856f929163a0aa20c26a78d26bfdc839f51b9d5aa7a5b79e52b7e83" +dependencies = [ + "rand 0.8.5", +] + [[package]] name = "num-bigint" version = "0.4.6" @@ -5770,12 +5834,12 @@ dependencies = [ [[package]] name = "redis" -version = "0.29.5" +version = "0.31.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1bc42f3a12fd4408ce64d8efef67048a924e543bd35c6591c0447fda9054695f" +checksum = "0bc1ea653e0b2e097db3ebb5b7f678be339620b8041f66b30a308c1d45d36a7f" dependencies = [ - "arc-swap", "bytes", + "cfg-if", "combine", "crc16", "futures-sink", @@ -6325,6 +6389,19 @@ dependencies = [ "security-framework 2.11.1", ] +[[package]] +name = "rustls-native-certs" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5bfb394eeed242e909609f56089eecfe5fda225042e8b171791b9c95f5931e5" +dependencies = [ + "openssl-probe", + "rustls-pemfile 2.2.0", + "rustls-pki-types", + "schannel", + "security-framework 2.11.1", +] + [[package]] name = "rustls-native-certs" version = "0.8.1" @@ -6670,6 +6747,15 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_nanos" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a93142f0367a4cc53ae0fead1bcda39e85beccfad3dcd717656cacab94b12985" +dependencies = [ + "serde", +] + [[package]] name = "serde_qs" version = "0.8.5" @@ -6681,6 +6767,17 @@ dependencies = [ "thiserror 1.0.69", ] +[[package]] +name = "serde_repr" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "175ee3e80ae9982737ca543e96133087cbd9a485eecc3bc4de9c1a37b47ea59c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.101", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -6883,6 +6980,18 @@ dependencies = [ "libc", ] +[[package]] +name = "signatory" +version = "0.27.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c1e303f8205714074f6068773f0e29527e0453937fe837c9717d066635b65f31" +dependencies = [ + "pkcs8", + "rand_core 0.6.4", + "signature", + "zeroize", +] + [[package]] name = "signature" version = "2.2.0" @@ -7108,6 +7217,7 @@ version = "0.11.8" dependencies = [ "ahash", "arc-swap", + "async-nats", "async-trait", "azure_core", "azure_storage", @@ -7628,6 +7738,27 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-websockets" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f591660438b3038dd04d16c938271c79e7e06260ad2ea2885a4861bfb238605d" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures-core", + "futures-sink", + "http 1.3.1", + "httparse", + "rand 0.8.5", + "ring 0.17.14", + "rustls-pki-types", + "tokio", + "tokio-rustls 0.26.2", + "tokio-util", + "webpki-roots 0.26.10", +] + [[package]] name = "toml_datetime" version = "0.6.9" @@ -7835,6 +7966,17 @@ version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fb1626d07cb5c1bb2cf17d94c0be4852e8a7c02b041acec9a8c5bdda99f9d580" +[[package]] +name = "tryhard" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c9f0a709784e86923586cff0d872dba54cd2d2e116b3bc57587d15737cfce9d" +dependencies = [ + "futures", + "pin-project-lite", + "tokio", +] + [[package]] name = "tungstenite" version = "0.21.0" diff --git a/crates/common/src/config/inner.rs b/crates/common/src/config/inner.rs index ce4f8177..66e55290 100644 --- a/crates/common/src/config/inner.rs +++ b/crates/common/src/config/inner.rs @@ -4,24 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::{ - net::{IpAddr, Ipv4Addr, Ipv6Addr}, - sync::Arc, -}; - -use ahash::{AHashMap, AHashSet}; -use arc_swap::ArcSwap; -use mail_auth::{MX, Parameters, Txt}; -use mail_send::smtp::tls::build_tls_connector; -use nlp::bayes::{TokenHash, Weights}; -use parking_lot::RwLock; -use store::write::BatchBuilder; -use utils::{ - cache::{Cache, CacheWithTtl}, - config::Config, - snowflake::SnowflakeIdGenerator, -}; - +use super::server::tls::{build_self_signed_cert, parse_certificates}; use crate::{ CacheSwap, Caches, Data, DavResource, DavResources, MailboxCache, MessageStoreCache, MessageUidCache, TlsConnectors, @@ -30,8 +13,21 @@ use crate::{ listener::blocked::BlockedIps, manager::webadmin::WebAdminManager, }; - -use super::server::tls::{build_self_signed_cert, parse_certificates}; +use ahash::{AHashMap, AHashSet}; +use arc_swap::ArcSwap; +use mail_auth::{MX, Parameters, Txt}; +use mail_send::smtp::tls::build_tls_connector; +use nlp::bayes::{TokenHash, Weights}; +use parking_lot::RwLock; +use std::{ + net::{IpAddr, Ipv4Addr, Ipv6Addr}, + sync::Arc, +}; +use utils::{ + cache::{Cache, CacheWithTtl}, + config::Config, + snowflake::{HlcTimestamp, SnowflakeIdGenerator}, +}; impl Data { pub fn parse(config: &mut Config) -> Self { @@ -48,7 +44,7 @@ impl Data { .property::("cluster.node-id") .unwrap_or_else(store::rand::random); let id_generator = SnowflakeIdGenerator::with_node_id(node_id); - BatchBuilder::init_id_generator(node_id as u16); + HlcTimestamp::init(node_id as u16); if !id_generator.is_valid() { panic!("Invalid system time, panicking to avoid data corruption"); } @@ -65,7 +61,6 @@ impl Data { .ok() .map(Arc::new), blocked_ips: RwLock::new(BlockedIps::parse(config).blocked_ip_addresses), - blocked_ips_version: 0.into(), jmap_id_gen: id_generator.clone(), queue_id_gen: id_generator.clone(), span_id_gen: id_generator, @@ -74,7 +69,6 @@ impl Data { .value("webadmin.path") .map(|path| WebAdminManager::new(path.into())) .unwrap_or_default(), - config_version: 0.into(), logos: Default::default(), smtp_connectors: TlsConnectors::default(), asn_geo_data: Default::default(), @@ -225,13 +219,11 @@ impl Default for Data { tls_certificates: Default::default(), tls_self_signed_cert: Default::default(), blocked_ips: Default::default(), - blocked_ips_version: 0.into(), jmap_id_gen: Default::default(), queue_id_gen: Default::default(), span_id_gen: Default::default(), queue_status: true.into(), webadmin: Default::default(), - config_version: Default::default(), logos: Default::default(), smtp_connectors: Default::default(), asn_geo_data: Default::default(), diff --git a/crates/common/src/config/mod.rs b/crates/common/src/config/mod.rs index 1cf4efb2..405838f2 100644 --- a/crates/common/src/config/mod.rs +++ b/crates/common/src/config/mod.rs @@ -137,6 +137,21 @@ impl Core { } }) .unwrap_or_default(); + let pubsub = config + .value("storage.pubsub") + .map(|id| id.to_string()) + .and_then(|id| { + if let Some(store) = stores.pubsub_stores.get(&id) { + store.clone().into() + } else { + config.new_parse_error( + "storage.pubsub", + format!("PubSub store {id:?} not found"), + ); + None + } + }) + .unwrap_or_default(); let mut directories = Directories::parse(config, &stores, data.clone(), is_enterprise).await; let directory = config @@ -192,6 +207,7 @@ impl Core { blob, fts, lookup, + pubsub, directory, directories: directories.directories, purge_schedules: stores.purge_schedules, diff --git a/crates/common/src/config/storage.rs b/crates/common/src/config/storage.rs index 246da4d6..46ba3efb 100644 --- a/crates/common/src/config/storage.rs +++ b/crates/common/src/config/storage.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use ahash::AHashMap; use directory::Directory; -use store::{BlobStore, FtsStore, InMemoryStore, PurgeSchedule, Store}; +use store::{BlobStore, FtsStore, InMemoryStore, PubSubStore, PurgeSchedule, Store}; use crate::manager::config::ConfigManager; @@ -18,6 +18,7 @@ pub struct Storage { pub blob: BlobStore, pub fts: FtsStore, pub lookup: InMemoryStore, + pub pubsub: PubSubStore, pub directory: Arc, pub directories: AHashMap>, pub purge_schedules: Vec, diff --git a/crates/common/src/core.rs b/crates/common/src/core.rs index 6d86bd88..47d58665 100644 --- a/crates/common/src/core.rs +++ b/crates/common/src/core.rs @@ -4,8 +4,15 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::{sync::Arc, time::Duration}; - +use crate::{ + Inner, Server, + auth::{AccessToken, ResourceToken, TenantInfo}, + config::smtp::{ + auth::{ArcSealer, DkimSigner, LazySignature, ResolvedSignature, build_signature}, + queue::RelayHost, + }, + ipc::{BroadcastEvent, StateEvent}, +}; use directory::{Directory, QueryBy, Type, backend::internal::manage::ManageDirectory}; use jmap_proto::types::{ blob::BlobId, @@ -15,6 +22,7 @@ use jmap_proto::types::{ type_state::DataType, }; use sieve::Sieve; +use std::{sync::Arc, time::Duration}; use store::{ BitmapKey, BlobClass, BlobStore, Deserialize, FtsStore, InMemoryStore, IndexKey, IterateParams, LogKey, SerializeInfallible, Store, U32_LEN, ValueKey, @@ -28,16 +36,6 @@ use store::{ use trc::AddContext; use utils::BlobHash; -use crate::{ - Inner, Server, - auth::{AccessToken, ResourceToken, TenantInfo}, - config::smtp::{ - auth::{ArcSealer, DkimSigner, LazySignature, ResolvedSignature, build_signature}, - queue::RelayHost, - }, - ipc::StateEvent, -}; - impl Server { #[inline(always)] pub fn store(&self) -> &Store { @@ -510,22 +508,21 @@ impl Server { if let Some(changes) = builder.changes() { for (account_id, changed_collections) in changes { - let mut state_change = StateChange::new(account_id); - let change_id = changed_collections.change_id; + let mut state_change = StateChange::new(account_id, changed_collections.change_id); for changed_collection in changed_collections.changed_containers { if let Some(data_type) = DataType::try_from_id(changed_collection, true) { - state_change.set_change(data_type, change_id); + state_change.set_change(data_type); } } for changed_collection in changed_collections.changed_items { if let Some(data_type) = DataType::try_from_id(changed_collection, false) { - state_change.set_change(data_type, change_id); + state_change.set_change(data_type); } } if state_change.has_changes() { self.broadcast_state_change(state_change).await; } - assigned_ids.change_id = change_id.into(); + assigned_ids.change_id = changed_collections.change_id.into(); } } @@ -576,7 +573,10 @@ impl Server { .ipc .state_tx .clone() - .send(StateEvent::Publish { state_change }) + .send(StateEvent::Publish { + state_change, + broadcast: true, + }) .await { Ok(_) => true, @@ -592,6 +592,18 @@ impl Server { } } + pub async fn cluster_broadcast(&self, event: BroadcastEvent) { + if let Some(broadcast_tx) = &self.inner.ipc.broadcast_tx.clone() { + if broadcast_tx.send(event).await.is_err() { + trc::event!( + Server(trc::ServerEvent::ThreadError), + Details = "Error sending broadcast event.", + CausedBy = trc::location!() + ); + } + } + } + #[allow(clippy::blocks_in_conditions)] pub async fn put_blob( &self, @@ -655,13 +667,6 @@ impl Server { }) } - pub fn increment_config_version(&self) { - self.inner - .data - .config_version - .fetch_add(1, std::sync::atomic::Ordering::Relaxed); - } - pub async fn total_accounts(&self) -> trc::Result { self.store() .count_principals(None, Type::Individual.into(), None) diff --git a/crates/common/src/ipc.rs b/crates/common/src/ipc.rs index 7853596e..cef27ef9 100644 --- a/crates/common/src/ipc.rs +++ b/crates/common/src/ipc.rs @@ -54,6 +54,7 @@ pub enum StateEvent { }, Publish { state_change: StateChange, + broadcast: bool, }, UpdateSharedAccounts { account_id: u32, @@ -65,6 +66,13 @@ pub enum StateEvent { Stop, } +#[derive(Debug)] +pub enum BroadcastEvent { + StateChange(StateChange), + ReloadSettings, + ReloadBlockedIps, +} + #[derive(Debug)] pub enum UpdateSubscription { Unverified { diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index ddfbcaf1..990e5cc6 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -24,7 +24,7 @@ use config::{ storage::Storage, telemetry::Metrics, }; -use ipc::{HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent}; +use ipc::{BroadcastEvent, HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent}; use jmap_proto::types::value::AclGrant; use listener::{asn::AsnGeoLookupData, blocked::Security, tls::AcmeProviders}; use mail_auth::{MX, Txt}; @@ -35,10 +35,7 @@ use rustls::sign::CertifiedKey; use std::{ hash::{BuildHasher, Hash, Hasher}, net::{IpAddr, Ipv4Addr, Ipv6Addr}, - sync::{ - Arc, - atomic::{AtomicBool, AtomicU8}, - }, + sync::{Arc, atomic::AtomicBool}, time::Duration, }; use store::roaring::RoaringBitmap; @@ -126,7 +123,6 @@ pub struct Data { pub tls_self_signed_cert: Option>, pub blocked_ips: RwLock>, - pub blocked_ips_version: AtomicU8, pub asn_geo_data: AsnGeoLookupData, @@ -137,7 +133,6 @@ pub struct Data { pub webadmin: WebAdminManager, pub logos: Mutex>>>>, - pub config_version: AtomicU8, pub smtp_connectors: TlsConnectors, } @@ -233,6 +228,7 @@ pub struct Ipc { pub index_tx: Arc, pub queue_tx: mpsc::Sender, pub report_tx: mpsc::Sender, + pub broadcast_tx: Option>, pub local_delivery_sm: Arc, } @@ -476,6 +472,7 @@ impl Default for Ipc { index_tx: Default::default(), queue_tx: mpsc::channel(IPC_CHANNEL_BUFFER).0, report_tx: mpsc::channel(IPC_CHANNEL_BUFFER).0, + broadcast_tx: None, local_delivery_sm: Arc::new(Semaphore::new(10)), } } diff --git a/crates/common/src/listener/blocked.rs b/crates/common/src/listener/blocked.rs index f92c92c0..564cd79c 100644 --- a/crates/common/src/listener/blocked.rs +++ b/crates/common/src/listener/blocked.rs @@ -18,7 +18,7 @@ use utils::{ use crate::{ KV_RATE_LIMIT_AUTH, KV_RATE_LIMIT_LOITER, KV_RATE_LIMIT_RCPT, KV_RATE_LIMIT_SCAN, Server, - ip_to_bytes, manager::config::MatchType, + ip_to_bytes, ipc::BroadcastEvent, manager::config::MatchType, }; #[derive(Debug, Clone)] @@ -226,7 +226,7 @@ impl Server { Ok(false) } - async fn block_ip(&self, ip: IpAddr) -> trc::Result<()> { + pub async fn block_ip(&self, ip: IpAddr) -> trc::Result<()> { // Add IP to blocked list self.inner.data.blocked_ips.write().insert(ip); @@ -244,7 +244,8 @@ impl Server { .await?; // Increment version - self.increment_blocked_version(); + self.cluster_broadcast(BroadcastEvent::ReloadBlockedIps) + .await; Ok(()) } @@ -276,13 +277,6 @@ impl Server { .iter() .any(|network| network.matches(ip))) } - - pub fn increment_blocked_version(&self) { - self.inner - .data - .blocked_ips_version - .fetch_add(1, std::sync::atomic::Ordering::Relaxed); - } } impl BlockedIps { diff --git a/crates/common/src/manager/boot.rs b/crates/common/src/manager/boot.rs index 92da2351..b0275bcd 100644 --- a/crates/common/src/manager/boot.rs +++ b/crates/common/src/manager/boot.rs @@ -27,7 +27,7 @@ use crate::{ Caches, Core, Data, IPC_CHANNEL_BUFFER, Inner, Ipc, config::{network::AsnGeoLookupConfig, server::Listeners, telemetry::Telemetry}, core::BuildServer, - ipc::{HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent}, + ipc::{BroadcastEvent, HousekeeperEvent, QueueEvent, ReportingEvent, StateEvent}, }; use super::{ @@ -49,6 +49,7 @@ pub struct IpcReceivers { pub housekeeper_rx: Option>, pub queue_rx: Option>, pub report_rx: Option>, + pub broadcast_rx: Option>, } const HELP: &str = concat!( @@ -424,7 +425,7 @@ impl BootManager { core.network.asn_geo_lookup, AsnGeoLookupConfig::Resource { .. } ); - let (ipc, ipc_rxs) = build_ipc(&mut config); + let (ipc, ipc_rxs) = build_ipc(&mut config, !core.storage.pubsub.is_none()); let inner = Arc::new(Inner { shared_core: ArcSwap::from_pointee(core), data, @@ -487,18 +488,20 @@ impl BootManager { } } -pub fn build_ipc(config: &mut Config) -> (Ipc, IpcReceivers) { +pub fn build_ipc(config: &mut Config, has_pubsub: bool) -> (Ipc, IpcReceivers) { // Build ipc receivers let (state_tx, state_rx) = mpsc::channel(IPC_CHANNEL_BUFFER); let (housekeeper_tx, housekeeper_rx) = mpsc::channel(IPC_CHANNEL_BUFFER); let (queue_tx, queue_rx) = mpsc::channel(IPC_CHANNEL_BUFFER); let (report_tx, report_rx) = mpsc::channel(IPC_CHANNEL_BUFFER); + let (broadcast_tx, broadcast_rx) = mpsc::channel(IPC_CHANNEL_BUFFER); ( Ipc { state_tx, housekeeper_tx, queue_tx, report_tx, + broadcast_tx: has_pubsub.then_some(broadcast_tx), index_tx: Arc::new(Notify::new()), local_delivery_sm: Arc::new(Semaphore::new( config @@ -512,6 +515,7 @@ pub fn build_ipc(config: &mut Config) -> (Ipc, IpcReceivers) { housekeeper_rx: Some(housekeeper_rx), queue_rx: Some(queue_rx), report_rx: Some(report_rx), + broadcast_rx: has_pubsub.then_some(broadcast_rx), }, ) } diff --git a/crates/common/src/manager/reload.rs b/crates/common/src/manager/reload.rs index 81810373..aeb12b46 100644 --- a/crates/common/src/manager/reload.rs +++ b/crates/common/src/manager/reload.rs @@ -76,6 +76,7 @@ impl Server { blob_stores: self.core.storage.blobs.clone(), fts_stores: self.core.storage.ftss.clone(), in_memory_stores: self.core.storage.lookups.clone(), + pubsub_stores: Default::default(), purge_schedules: Default::default(), }; stores.parse_stores(&mut config).await; diff --git a/crates/email/src/message/delivery.rs b/crates/email/src/message/delivery.rs index 9e2b3c5e..6d848d32 100644 --- a/crates/email/src/message/delivery.rs +++ b/crates/email/src/message/delivery.rs @@ -213,11 +213,11 @@ impl MailDelivery for Server { // Notify state change if ingested_message.change_id != u64::MAX { self.broadcast_state_change( - StateChange::new(uid) - .with_change(DataType::EmailDelivery, ingested_message.change_id) - .with_change(DataType::Email, ingested_message.change_id) - .with_change(DataType::Mailbox, ingested_message.change_id) - .with_change(DataType::Thread, ingested_message.change_id), + StateChange::new(uid, ingested_message.change_id) + .with_change(DataType::EmailDelivery) + .with_change(DataType::Email) + .with_change(DataType::Mailbox) + .with_change(DataType::Thread), ) .await; } diff --git a/crates/http/src/management/reload.rs b/crates/http/src/management/reload.rs index aec61963..57f23fc6 100644 --- a/crates/http/src/management/reload.rs +++ b/crates/http/src/management/reload.rs @@ -4,7 +4,11 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use common::{Server, auth::AccessToken, ipc::HousekeeperEvent}; +use common::{ + Server, + auth::AccessToken, + ipc::{BroadcastEvent, HousekeeperEvent}, +}; use directory::Permission; use hyper::Method; use serde_json::json; @@ -59,8 +63,8 @@ impl ManageReload for Server { (Some("server.blocked-ip"), &Method::GET) => { let result = self.reload_blocked_ips().await?; - // Increment version counter - self.increment_blocked_version(); + self.cluster_broadcast(BroadcastEvent::ReloadBlockedIps) + .await; Ok(JsonResponse::new(json!({ "data": result.config, @@ -74,8 +78,7 @@ impl ManageReload for Server { // Update core self.inner.shared_core.store(core.into()); - // Increment version counter - self.increment_config_version(); + self.cluster_broadcast(BroadcastEvent::ReloadSettings).await; } if let Some(tracers) = result.tracers { diff --git a/crates/imap/src/op/append.rs b/crates/imap/src/op/append.rs index b3d46582..46158077 100644 --- a/crates/imap/src/op/append.rs +++ b/crates/imap/src/op/append.rs @@ -145,10 +145,10 @@ impl SessionData { if let Some(change_id) = last_change_id { self.server .broadcast_state_change( - StateChange::new(account_id) - .with_change(DataType::Email, change_id) - .with_change(DataType::Mailbox, change_id) - .with_change(DataType::Thread, change_id), + StateChange::new(account_id, change_id) + .with_change(DataType::Email) + .with_change(DataType::Mailbox) + .with_change(DataType::Thread), ) .await; } diff --git a/crates/imap/src/op/copy_move.rs b/crates/imap/src/op/copy_move.rs index 1f4b0780..f490fd1f 100644 --- a/crates/imap/src/op/copy_move.rs +++ b/crates/imap/src/op/copy_move.rs @@ -368,10 +368,10 @@ impl SessionData { if let Some(change_id) = dest_change_id { self.server .broadcast_state_change( - StateChange::new(dest_account_id) - .with_change(DataType::Email, change_id) - .with_change(DataType::Thread, change_id) - .with_change(DataType::Mailbox, change_id), + StateChange::new(dest_account_id, change_id) + .with_change(DataType::Email) + .with_change(DataType::Thread) + .with_change(DataType::Mailbox), ) .await; } diff --git a/crates/imap/src/op/idle.rs b/crates/imap/src/op/idle.rs index 55785095..1c265b86 100644 --- a/crates/imap/src/op/idle.rs +++ b/crates/imap/src/op/idle.rs @@ -96,7 +96,7 @@ impl Session { let mut has_mailbox_changes = false; let mut has_email_changes = false; - for (type_state, _) in state_change.types { + for type_state in state_change.types { match type_state { DataType::Email | DataType::EmailDelivery => { has_email_changes = true; diff --git a/crates/jmap-proto/src/types/state.rs b/crates/jmap-proto/src/types/state.rs index a19160d2..f04d61c1 100644 --- a/crates/jmap-proto/src/types/state.rs +++ b/crates/jmap-proto/src/types/state.rs @@ -4,9 +4,12 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use utils::codec::{ - base32_custom::Base32Writer, - leb128::{Leb128Iterator, Leb128Writer}, +use utils::{ + codec::{ + base32_custom::Base32Writer, + leb128::{Leb128Iterator, Leb128Writer}, + }, + map::bitmap::Bitmap, }; use crate::parser::{JsonObjectParser, base32::JsonBase32Reader, json::Parser}; @@ -28,30 +31,28 @@ pub enum State { Intermediate(JMAPIntermediateState), } -#[derive(Clone, Debug)] +#[derive(Debug, Clone, Copy)] pub struct StateChange { pub account_id: u32, - pub types: Vec<(DataType, u64)>, + pub change_id: u64, + pub types: Bitmap, } impl StateChange { - pub fn new(account_id: u32) -> Self { + pub fn new(account_id: u32, change_id: u64) -> Self { Self { account_id, - types: Vec::with_capacity(0), + change_id, + types: Default::default(), } } - pub fn set_change(&mut self, type_state: DataType, change_id: u64) { - self.types.push((type_state, change_id)); + pub fn set_change(&mut self, type_state: DataType) { + self.types.insert(type_state); } - pub fn with_change(mut self, type_state: DataType, change_id: u64) -> Self { - if let Some((_, last_change_id)) = self.types.iter_mut().find(|(ts, _)| ts == &type_state) { - *last_change_id = change_id; - } else { - self.types.push((type_state, change_id)); - } + pub fn with_change(mut self, type_state: DataType) -> Self { + self.set_change(type_state); self } diff --git a/crates/jmap/src/api/event_source.rs b/crates/jmap/src/api/event_source.rs index d26621f7..f9380913 100644 --- a/crates/jmap/src/api/event_source.rs +++ b/crates/jmap/src/api/event_source.rs @@ -116,11 +116,11 @@ impl EventSourceHandler for Server { loop { match tokio::time::timeout(timeout, change_rx.recv()).await { Ok(Some(state_change)) => { - for (type_state, change_id) in state_change.types { + for type_state in state_change.types { response .changed .get_mut_or_insert(state_change.account_id.into()) - .set(type_state, change_id.into()); + .set(type_state, state_change.change_id.into()); } } Ok(None) => { diff --git a/crates/jmap/src/email/set.rs b/crates/jmap/src/email/set.rs index 1db2182f..cd396ee4 100644 --- a/crates/jmap/src/email/set.rs +++ b/crates/jmap/src/email/set.rs @@ -1040,10 +1040,10 @@ impl EmailSet for Server { if response.updated.is_empty() && response.destroyed.is_empty() { // Message ingest does not broadcast state changes self.broadcast_state_change( - StateChange::new(account_id) - .with_change(DataType::Email, change_id) - .with_change(DataType::Mailbox, change_id) - .with_change(DataType::Thread, change_id), + StateChange::new(account_id, change_id) + .with_change(DataType::Email) + .with_change(DataType::Mailbox) + .with_change(DataType::Thread), ) .await; } diff --git a/crates/jmap/src/websocket/stream.rs b/crates/jmap/src/websocket/stream.rs index 3a37d85d..42347773 100644 --- a/crates/jmap/src/websocket/stream.rs +++ b/crates/jmap/src/websocket/stream.rs @@ -178,18 +178,15 @@ impl WebSocketHandler for Server { } state_change = change_rx.recv() => { if let Some(state_change) = state_change { - if !change_types.is_empty() && state_change - .types - .iter() - .any(|(t, _)| change_types.contains(*t)) - { - for (type_state, change_id) in state_change.types { - changes - .changed - .get_mut_or_insert(state_change.account_id.into()) - .set(type_state, change_id.into()); - } - } + let mut types = state_change.types; + types.intersection(&change_types); + + for type_state in types { + changes + .changed + .get_mut_or_insert(state_change.account_id.into()) + .set(type_state, state_change.change_id.into()); + } } else { trc::event!( Jmap(JmapEvent::WebsocketStop), diff --git a/crates/main/Cargo.toml b/crates/main/Cargo.toml index 4ee570ae..0af9edcb 100644 --- a/crates/main/Cargo.toml +++ b/crates/main/Cargo.toml @@ -39,7 +39,7 @@ tokio = { version = "1.23", features = ["full"] } jemallocator = "0.5.0" [features] -#default = ["sqlite", "postgres", "mysql", "rocks", "elastic", "s3", "redis", "azure", "enterprise"] +#default = ["sqlite", "postgres", "mysql", "rocks", "elastic", "s3", "redis", "azure", "nats", "enterprise"] default = ["rocks", "enterprise"] sqlite = ["store/sqlite"] foundationdb = ["store/foundation", "common/foundation"] @@ -49,6 +49,7 @@ rocks = ["store/rocks"] elastic = ["store/elastic"] s3 = ["store/s3"] redis = ["store/redis"] +nats = ["store/nats"] azure = ["store/azure"] enterprise = [ "jmap/enterprise", "smtp/enterprise", diff --git a/crates/main/src/main.rs b/crates/main/src/main.rs index bbbceb5a..870d59e0 100644 --- a/crates/main/src/main.rs +++ b/crates/main/src/main.rs @@ -13,7 +13,7 @@ use http::HttpSessionManager; use imap::core::ImapSessionManager; use managesieve::core::ManageSieveSessionManager; use pop3::Pop3SessionManager; -use services::{StartServices, gossip::spawn::GossiperBuilder}; +use services::{StartServices, broadcast::subscriber::spawn_broadcast_subscriber}; use smtp::{StartQueueManager, core::SmtpSessionManager}; use trc::Collector; use utils::wait_for_shutdown; @@ -33,7 +33,6 @@ async fn main() -> std::io::Result<()> { // Init services init.start_services().await; init.start_queue_manager(); - let gossiper = GossiperBuilder::try_parse(&mut init.config); // Log configuration errors init.config.log_errors(); @@ -83,10 +82,8 @@ async fn main() -> std::io::Result<()> { }; }); - // Spawn gossip - if let Some(gossiper) = gossiper { - gossiper.spawn(init.inner, shutdown_rx.clone()).await; - } + // Start broadcast subscriber + spawn_broadcast_subscriber(init.inner, shutdown_rx); // Wait for shutdown signal wait_for_shutdown().await; diff --git a/crates/services/src/broadcast/mod.rs b/crates/services/src/broadcast/mod.rs new file mode 100644 index 00000000..b1410bf7 --- /dev/null +++ b/crates/services/src/broadcast/mod.rs @@ -0,0 +1,113 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use common::ipc::BroadcastEvent; +use jmap_proto::types::state::StateChange; +use utils::map::bitmap::Bitmap; + +pub mod publisher; +pub mod subscriber; + +#[derive(Debug)] +pub(crate) struct BroadcastBatch { + messages: T, +} + +const MAX_BATCH_SIZE: usize = 100; +const MESSAGE_SIZE: usize = std::mem::size_of::() + (std::mem::size_of::() * 2); +pub(crate) const BROADCAST_TOPIC: &str = "stwt.agora"; + +impl BroadcastBatch> { + pub fn init() -> Self { + Self { + messages: Vec::with_capacity(MAX_BATCH_SIZE), + } + } + + pub fn insert(&mut self, message: BroadcastEvent) -> bool { + self.messages.push(message); + self.messages.len() < MAX_BATCH_SIZE + } + + pub fn serialize(&self, node_id: u16) -> Vec { + let mut serialized = + Vec::with_capacity((self.messages.len() * MESSAGE_SIZE) + std::mem::size_of::()); + serialized.extend_from_slice(&node_id.to_le_bytes()); + for message in &self.messages { + let msg_id: u32 = match message { + BroadcastEvent::StateChange(state_change) => { + serialized.extend_from_slice(&state_change.change_id.to_le_bytes()); + serialized.extend_from_slice(&state_change.types.as_ref().to_le_bytes()); + serialized.extend_from_slice(&state_change.account_id.to_le_bytes()); + continue; + } + BroadcastEvent::ReloadSettings => 0, + BroadcastEvent::ReloadBlockedIps => 1, + }; + + serialized.extend_from_slice(&u64::MAX.to_le_bytes()); + serialized.extend_from_slice(&u64::MAX.to_le_bytes()); + serialized.extend_from_slice(&msg_id.to_le_bytes()); + } + serialized + } + + pub fn clear(&mut self) { + self.messages.clear(); + } +} + +impl> BroadcastBatch { + pub fn node_id(&self) -> Option { + self.messages + .as_ref() + .get(0..std::mem::size_of::()) + .and_then(|bytes| bytes.try_into().ok()) + .map(u16::from_le_bytes) + } + + pub fn events(&self) -> impl Iterator> { + self.messages + .as_ref() + .get(std::mem::size_of::()..) + .unwrap_or_default() + .chunks_exact(MESSAGE_SIZE) + .map(|chunk| { + let change_id = + u64::from_le_bytes(chunk[0..std::mem::size_of::()].try_into().unwrap()); + let types = u64::from_le_bytes( + chunk[std::mem::size_of::()..std::mem::size_of::() * 2] + .try_into() + .unwrap(), + ); + let account_id = u32::from_le_bytes( + chunk[std::mem::size_of::() * 2..20] + .try_into() + .unwrap(), + ); + + Some(if change_id != u64::MAX { + BroadcastEvent::StateChange(StateChange { + change_id, + types: Bitmap::from(types), + account_id, + }) + } else { + match account_id { + 0 => BroadcastEvent::ReloadSettings, + 1 => BroadcastEvent::ReloadBlockedIps, + _ => return None, + } + }) + }) + } +} + +impl BroadcastBatch { + pub fn new(messages: T) -> Self { + Self { messages } + } +} diff --git a/crates/services/src/broadcast/publisher.rs b/crates/services/src/broadcast/publisher.rs new file mode 100644 index 00000000..b62be3b0 --- /dev/null +++ b/crates/services/src/broadcast/publisher.rs @@ -0,0 +1,55 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use std::sync::Arc; + +use common::{Inner, ipc::BroadcastEvent}; +use tokio::sync::mpsc; +use trc::ClusterEvent; + +use super::{BROADCAST_TOPIC, BroadcastBatch}; + +pub fn spawn_broadcast_publisher(inner: Arc, mut event_rx: mpsc::Receiver) { + let (pubsub, this_node_id) = { + let _core = inner.shared_core.load(); + let pubsub = inner.shared_core.load().storage.pubsub.clone(); + if pubsub.is_none() { + return; + } + (pubsub, _core.network.node_id as u16) + }; + + tokio::spawn(async move { + let mut batch = BroadcastBatch::init(); + + trc::event!(Cluster(ClusterEvent::PublisherStart)); + + while let Some(event) = event_rx.recv().await { + batch.insert(event); + + while let Ok(event) = event_rx.try_recv() { + if !batch.insert(event) { + break; + } + } + + match pubsub + .publish(BROADCAST_TOPIC, batch.serialize(this_node_id)) + .await + { + Ok(_) => { + batch.clear(); + } + Err(err) => { + batch.clear(); + trc::event!(Cluster(ClusterEvent::PublisherError), CausedBy = err); + } + } + } + + trc::event!(Cluster(ClusterEvent::PublisherStop)); + }); +} diff --git a/crates/services/src/broadcast/subscriber.rs b/crates/services/src/broadcast/subscriber.rs new file mode 100644 index 00000000..7a058ec4 --- /dev/null +++ b/crates/services/src/broadcast/subscriber.rs @@ -0,0 +1,222 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::broadcast::{BROADCAST_TOPIC, BroadcastBatch}; +use common::{ + Inner, + core::BuildServer, + ipc::{BroadcastEvent, HousekeeperEvent, StateEvent}, +}; +use compact_str::CompactString; +use std::{sync::Arc, time::Duration}; +use tokio::sync::watch; +use trc::{ClusterEvent, ServerEvent}; +use utils::snowflake::HlcTimestamp; + +pub fn spawn_broadcast_subscriber(inner: Arc, mut shutdown_rx: watch::Receiver) { + let this_node_id = { + let _core = inner.shared_core.load(); + if _core.storage.pubsub.is_none() { + return; + } + _core.network.node_id as u16 + }; + + tokio::spawn(async move { + let mut retry_count = 0; + + trc::event!(Cluster(ClusterEvent::SubscriberStart)); + + loop { + let pubsub = inner.shared_core.load().storage.pubsub.clone(); + if pubsub.is_none() { + trc::event!( + Cluster(ClusterEvent::SubscriberError), + Details = "PubSub is no longer configured" + ); + break; + } + + let mut stream = match pubsub.subscribe(BROADCAST_TOPIC).await { + Ok(stream) => { + retry_count = 0; + stream + } + Err(err) => { + trc::event!( + Cluster(ClusterEvent::SubscriberError), + CausedBy = err, + Details = "Failed to subscribe to channel" + ); + + match tokio::time::timeout( + Duration::from_secs(1 << retry_count.clamp(1, 6)), + shutdown_rx.changed(), + ) + .await + { + Ok(_) => { + break; + } + Err(_) => { + retry_count += 1; + + continue; + } + } + } + }; + + tokio::select! { + message = stream.next() => { + match message { + Some(message) => { + let batch = BroadcastBatch::new(message.payload()); + let node_id = match batch.node_id() { + Some(node_id) => { + if node_id != this_node_id { + node_id + } else { + trc::event!( + Cluster(ClusterEvent::MessageSkipped), + Details = message.payload() + ); + continue; + } + } + None => { + trc::event!( + Cluster(ClusterEvent::MessageInvalid), + Details = message.payload() + ); + continue; + } + }; + + let mut max_timestamp = 0; + let mut num_events = 0; + let mut has_errors = false; + + for event in batch.events() { + if let Some(event) = event { + match event { + BroadcastEvent::StateChange(state_change) => { + max_timestamp = std::cmp::max( + max_timestamp, + state_change.change_id, + ); + if inner.ipc.state_tx.send(StateEvent::Publish { state_change, broadcast: false }).await.is_err() { + trc::event!( + Server(ServerEvent::ThreadError), + Details = "Error sending state change.", + CausedBy = trc::location!() + ); + } + }, + BroadcastEvent::ReloadSettings => { + match inner.build_server().reload().await { + Ok(result) => { + if let Some(new_core) = result.new_core { + // Update core + inner.shared_core.store(new_core.into()); + + if inner + .ipc + .housekeeper_tx + .send(HousekeeperEvent::ReloadSettings) + .await + .is_err() + { + trc::event!( + Server(trc::ServerEvent::ThreadError), + Details = "Failed to send setting reload event to housekeeper", + CausedBy = trc::location!(), + ); + } + } + } + Err(err) => { + trc::error!( + err.details("Failed to reload settings") + .caused_by(trc::location!()) + ); + } + } + }, + BroadcastEvent::ReloadBlockedIps => { + if let Err(err) = inner.build_server().reload_blocked_ips().await { + trc::error!( + err.details("Failed to reload settings") + .caused_by(trc::location!()) + ); + } + }, + } + num_events += 1; + } else if !has_errors { + trc::event!( + Cluster(ClusterEvent::MessageInvalid), + Details = message.payload() + ); + has_errors = true; + } + + } + + let skew = if max_timestamp != 0 { + match HlcTimestamp::update_clock_from_remote_timestamp(max_timestamp) { + Ok(skew) => Some(skew), + Err(large_skew) => { + trc::event!( + Cluster(ClusterEvent::ClockSkewDetected), + From = node_id, + To = this_node_id, + Total = num_events, + Details = large_skew, + ); + continue; + }, + } + } else { + None + }; + + trc::event!( + Cluster(ClusterEvent::MessageReceived), + From = node_id, + To = this_node_id, + Total = batch.events().flatten().map(log_event).collect::>(), + Details = skew, + ); + } + None => { + trc::event!( + Cluster(ClusterEvent::SubscriberDisconnected), + ); + } + } + }, + _ = shutdown_rx.changed() => { + break; + } + }; + } + + trc::event!(Cluster(ClusterEvent::SubscriberStop)); + }); +} + +fn log_event(event: BroadcastEvent) -> trc::Value { + match event { + BroadcastEvent::StateChange(state_change) => trc::Value::Array(vec![ + state_change.account_id.into(), + state_change.change_id.into(), + (*state_change.types.as_ref()).into(), + ]), + BroadcastEvent::ReloadSettings => CompactString::const_new("ReloadSettings").into(), + BroadcastEvent::ReloadBlockedIps => CompactString::const_new("ReloadBlockedIps").into(), + } +} diff --git a/crates/services/src/gossip/heartbeat.rs b/crates/services/src/gossip/heartbeat.rs deleted file mode 100644 index 01772908..00000000 --- a/crates/services/src/gossip/heartbeat.rs +++ /dev/null @@ -1,122 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -use trc::ClusterEvent; - -use super::{HEARTBEAT_WINDOW, HEARTBEAT_WINDOW_MASK, Peer, State}; -use std::time::Instant; - -// Phi Accrual Failure Detector defaults -const HB_MAX_PAUSE_MS: f64 = 0.0; -const HB_MIN_STD_DEV: f64 = 300.0; -const HB_PHI_SUSPECT_THRESHOLD: f64 = 5.0; -const HB_PHI_CONVICT_THRESHOLD: f64 = 9.0; - -impl Peer { - pub fn update_heartbeat(&mut self, is_direct_ping: bool) -> bool { - let hb_diff = - std::cmp::min(self.last_heartbeat.elapsed().as_millis(), 60 * 60 * 1000) as u64; - self.last_heartbeat = Instant::now(); - - match self.state { - State::Seed | State::Offline => { - trc::event!(Cluster(ClusterEvent::PeerAlive), RemoteIp = self.addr); - - self.state = State::Alive; - - // Do not count stale heartbeats. - return true; - } - State::Suspected => { - trc::event!( - Cluster(ClusterEvent::PeerSuspectedIsAlive), - RemoteIp = self.addr - ); - - self.state = State::Alive; - } - State::Left if is_direct_ping => { - trc::event!(Cluster(ClusterEvent::PeerBackOnline), RemoteIp = self.addr); - - self.state = State::Alive; - - // Do not count stale heartbeats. - return true; - } - _ => (), - } - - self.hb_window_pos = (self.hb_window_pos + 1) & HEARTBEAT_WINDOW_MASK; - - if !self.hb_is_full && self.hb_window_pos == 0 && self.hb_sum > 0 { - self.hb_is_full = true; - } - - if self.hb_is_full { - let hb_window = self.hb_window[self.hb_window_pos] as u64; - self.hb_sum -= hb_window; - self.hb_sq_sum -= hb_window.saturating_mul(hb_window); - } - - self.hb_window[self.hb_window_pos] = hb_diff as u32; - self.hb_sum += hb_diff; - self.hb_sq_sum += hb_diff.saturating_mul(hb_diff); - - false - } - - /* - Phi Accrual Failure Detection - Ported from https://github.com/akka/akka/blob/main/akka-remote/src/main/scala/akka/remote/PhiAccrualFailureDetector.scala - */ - pub fn check_heartbeat(&mut self) -> bool { - if self.hb_sum == 0 { - return false; - } - - let hb_diff = self.last_heartbeat.elapsed().as_millis() as f64; - let sample_size = if self.hb_is_full { - HEARTBEAT_WINDOW - } else { - self.hb_window_pos + 1 - } as f64; - let hb_mean = (self.hb_sum as f64 / sample_size) + HB_MAX_PAUSE_MS; - let hb_variance = (self.hb_sq_sum as f64 / sample_size) - (hb_mean * hb_mean); - let hb_std_dev = hb_variance.sqrt(); - let y = (hb_diff - hb_mean) / hb_std_dev.max(HB_MIN_STD_DEV); - let e = (-y * (1.5976 + 0.070566 * y * y)).exp(); - let phi = if hb_diff > hb_mean { - -(e / (1.0 + e)).log10() - } else { - -(1.0 - 1.0 / (1.0 + e)).log10() - }; - - /*trc::event!( - "Heartbeat from {}: mean={:.2}ms, variance={:.2}ms, std_dev={:.2}ms, phi={:.2}, samples={}, status={:?}", - self.addr, hb_mean, hb_variance, hb_std_dev, phi, sample_size, if phi > HB_PHI_CONVICT_THRESHOLD { - State::Offline - } else if phi > HB_PHI_SUSPECT_THRESHOLD { - State::Suspected - } else { - State::Alive - } - );*/ - - if phi > HB_PHI_CONVICT_THRESHOLD { - trc::event!(Cluster(ClusterEvent::PeerOffline), RemoteIp = self.addr); - - self.state = State::Offline; - false - } else if phi > HB_PHI_SUSPECT_THRESHOLD { - trc::event!(Cluster(ClusterEvent::PeerSuspected), RemoteIp = self.addr); - - self.state = State::Suspected; - true - } else { - true - } - } -} diff --git a/crates/services/src/gossip/leave.rs b/crates/services/src/gossip/leave.rs deleted file mode 100644 index fa124330..00000000 --- a/crates/services/src/gossip/leave.rs +++ /dev/null @@ -1,43 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -use trc::ClusterEvent; - -use crate::gossip::State; - -use super::request::Request; -use super::{Gossiper, PeerStatus}; - -impl Gossiper { - pub async fn broadcast_leave(&self) { - let mut status: Vec = Vec::with_capacity(self.peers.len() + 1); - status.push(self.into()); - for peer in &self.peers { - if !peer.is_offline() { - self.send_gossip(peer.addr, Request::Leave(status.clone())) - .await; - } - } - } - - pub async fn handle_leave(&mut self, peers: Vec) { - if let Some(peer) = peers.first() { - for local_peer in self.peers.iter_mut() { - if local_peer.addr == peer.addr { - trc::event!(Cluster(ClusterEvent::PeerLeaving), RemoteIp = peer.addr); - - local_peer.state = State::Left; - local_peer.epoch = peer.epoch; - - // Reload - self.request_reload(); - - break; - } - } - } - } -} diff --git a/crates/services/src/gossip/mod.rs b/crates/services/src/gossip/mod.rs deleted file mode 100644 index 10ff5d8e..00000000 --- a/crates/services/src/gossip/mod.rs +++ /dev/null @@ -1,127 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -pub mod heartbeat; -pub mod leave; -pub mod peer; -pub mod ping; -pub mod request; -pub mod spawn; - -use common::Inner; -use std::{ - net::{IpAddr, SocketAddr}, - sync::{Arc, atomic::Ordering}, - time::Instant, -}; -use tokio::sync::mpsc; -use trc::ClusterEvent; - -use self::request::Request; - -const UDP_MAX_PAYLOAD: usize = 65500; -const HEARTBEAT_WINDOW: usize = 1 << 10; -const HEARTBEAT_WINDOW_MASK: usize = HEARTBEAT_WINDOW - 1; - -pub type EpochId = u64; -pub type GenerationId = u8; - -pub struct Gossiper { - // Local node peer and shard id - pub addr: IpAddr, - pub port: u16, - - // Gossip state - pub epoch: EpochId, - - // Peer list - pub peers: Vec, - pub last_peer_pinged: usize, - - // IPC - pub inner: Arc, - pub gossip_tx: mpsc::Sender<(SocketAddr, Request)>, -} - -#[derive(Debug, Copy, Clone, PartialEq, Eq)] -pub enum State { - Seed, - Alive, - Suspected, - Offline, - Left, -} - -#[derive(Debug)] -pub struct Peer { - // Peer identity - pub addr: IpAddr, - - // Peer status - pub epoch: EpochId, - pub gen_config: GenerationId, - pub gen_lists: GenerationId, - pub state: State, - - // Heartbeat state - pub last_heartbeat: Instant, - pub hb_window: Vec, - pub hb_window_pos: usize, - pub hb_sum: u64, - pub hb_sq_sum: u64, - pub hb_is_full: bool, -} - -#[derive(Debug, Clone)] -pub struct PeerStatus { - pub addr: IpAddr, - pub epoch: EpochId, - pub gen_config: GenerationId, - pub gen_lists: GenerationId, -} - -impl From<&Peer> for PeerStatus { - fn from(peer: &Peer) -> Self { - PeerStatus { - addr: peer.addr, - epoch: peer.epoch, - gen_config: peer.gen_config, - gen_lists: peer.gen_lists, - } - } -} - -impl From<&Gossiper> for PeerStatus { - fn from(cluster: &Gossiper) -> Self { - PeerStatus { - addr: cluster.addr, - epoch: cluster.epoch, - gen_config: cluster.inner.data.config_version.load(Ordering::Relaxed), - gen_lists: cluster - .inner - .data - .blocked_ips_version - .load(Ordering::Relaxed), - } - } -} - -impl Gossiper { - async fn send_gossip(&self, dest: IpAddr, request: Request) { - if let Err(err) = self - .gossip_tx - .send((SocketAddr::new(dest, self.port), request)) - .await - { - trc::event!( - Cluster(ClusterEvent::Error), - RemoteIp = dest, - Details = "Failed to send gossip message", - Reason = err.to_string() - ); - }; - } -} diff --git a/crates/services/src/gossip/peer.rs b/crates/services/src/gossip/peer.rs deleted file mode 100644 index 66dd41af..00000000 --- a/crates/services/src/gossip/peer.rs +++ /dev/null @@ -1,98 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -use std::{fmt::Display, net::IpAddr, time::Instant}; - -use super::{Gossiper, HEARTBEAT_WINDOW, Peer, PeerStatus, State}; - -impl Peer { - pub fn new_seed(addr: IpAddr) -> Self { - Peer { - epoch: 0, - gen_config: 0, - gen_lists: 0, - addr, - state: State::Seed, - last_heartbeat: Instant::now(), - hb_window: vec![0; HEARTBEAT_WINDOW], - hb_window_pos: 0, - hb_sum: 0, - hb_sq_sum: 0, - hb_is_full: false, - } - } - - pub fn is_seed(&self) -> bool { - self.state == State::Seed - } - - pub fn is_alive(&self) -> bool { - self.state == State::Alive - } - - pub fn is_suspected(&self) -> bool { - self.state == State::Suspected - } - - pub fn is_healthy(&self) -> bool { - matches!(self.state, State::Alive | State::Suspected) - } - - pub fn is_offline(&self) -> bool { - matches!(self.state, State::Offline | State::Left) - } -} - -impl Gossiper { - pub fn is_peer_healthy(&self, addr: &IpAddr) -> bool { - self.peers.iter().any(|p| &p.addr == addr && p.is_healthy()) - } - - pub fn get_peer(&self, addr: &IpAddr) -> Option<&Peer> { - self.peers.iter().find(|p| &p.addr == addr) - } - - pub fn is_known_peer(&self, addr: &IpAddr) -> bool { - self.peers.iter().any(|p| &p.addr == addr) - } - - pub fn get_peer_mut(&mut self, addr: &IpAddr) -> Option<&mut Peer> { - self.peers.iter_mut().find(|p| &p.addr == addr) - } - - pub fn build_peer_status(&self) -> Vec { - let mut result: Vec = Vec::with_capacity(self.peers.len() + 1); - result.push(self.into()); - for peer in self.peers.iter() { - result.push(peer.into()); - } - result - } -} - -impl From for Peer { - fn from(value: PeerStatus) -> Self { - Peer { - addr: value.addr, - epoch: value.epoch, - gen_config: value.gen_config, - gen_lists: value.gen_lists, - state: State::Alive, - last_heartbeat: Instant::now(), - hb_window: vec![0; HEARTBEAT_WINDOW], - hb_window_pos: 0, - hb_sum: 0, - hb_sq_sum: 0, - hb_is_full: false, - } - } -} - -impl Display for Peer { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.addr) - } -} diff --git a/crates/services/src/gossip/ping.rs b/crates/services/src/gossip/ping.rs deleted file mode 100644 index 5764f7bc..00000000 --- a/crates/services/src/gossip/ping.rs +++ /dev/null @@ -1,206 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -use common::{ - core::BuildServer, - ipc::{HousekeeperEvent, QueueEvent}, -}; -use trc::ClusterEvent; - -use super::{Gossiper, PeerStatus, request::Request}; - -impl Gossiper { - pub async fn ping_peers(&mut self) { - // Total and alive peers in the cluster. - let total_peers = self.peers.len(); - let mut alive_peers: u32 = 0; - let mut node_became_offline = false; - - // Count alive peers - for peer in self.peers.iter_mut() { - if !peer.is_offline() { - if peer.check_heartbeat() { - alive_peers += 1; - } else if peer.hb_sum > 0 { - // Node is suspected to be offline - node_became_offline = true; - } - } - } - - // Find next peer to ping - for _ in 0..total_peers { - self.last_peer_pinged = (self.last_peer_pinged + 1) % total_peers; - let (peer_state, target_addr) = { - let peer = &self.peers[self.last_peer_pinged]; - (peer.state, peer.addr) - }; - - match peer_state { - super::State::Seed => { - self.send_gossip(target_addr, Request::Ping(vec![(&*self).into()])) - .await; - break; - } - super::State::Alive | super::State::Suspected => { - self.epoch += 1; - self.send_gossip(target_addr, Request::Ping(self.build_peer_status())) - .await; - break; - } - super::State::Offline if alive_peers == 0 => { - // Probe offline nodes - self.send_gossip(target_addr, Request::Ping(self.build_peer_status())) - .await; - break; - } - _ => (), - } - } - - if node_became_offline { - self.request_reload(); - } - } - - pub fn request_reload(&self) { - let server = self.inner.build_server(); - - tokio::spawn(async move { - trc::event!(Cluster(ClusterEvent::OneOrMorePeersOffline)); - - server.notify_task_queue(); - let _ = server.inner.ipc.queue_tx.send(QueueEvent::Refresh).await; - }); - } - - pub async fn broadcast_ping(&self) { - let status = self.build_peer_status(); - for peer in &self.peers { - if !peer.is_offline() { - self.send_gossip(peer.addr, Request::Pong(status.clone())) - .await; - } - } - } - - pub async fn handle_ping(&mut self, peers: Vec, send_pong: bool) { - // Increase epoch - self.epoch += 1; - - if peers.is_empty() { - trc::event!(Cluster(ClusterEvent::EmptyPacket)); - - return; - } - - let mut remove_seeds = false; - let mut update_config = false; - let mut update_lists = false; - - 'outer: for (pos, peer) in peers.into_iter().enumerate() { - if peer.addr == self.addr { - continue; - } - - for local_peer in self.peers.iter_mut() { - if !local_peer.is_seed() { - if local_peer.addr == peer.addr { - if peer.epoch > local_peer.epoch || pos == 0 { - local_peer.update_heartbeat(pos == 0); - local_peer.epoch = peer.epoch; - local_peer.addr = peer.addr; - if local_peer.gen_config != peer.gen_config { - local_peer.gen_config = peer.gen_config; - if local_peer.hb_sum > 0 { - trc::event!( - Cluster(ClusterEvent::PeerHasChanges), - RemoteIp = peer.addr, - Details = "settings" - ); - - update_config = true; - } - } - if local_peer.gen_lists != peer.gen_lists { - local_peer.gen_lists = peer.gen_lists; - if local_peer.hb_sum > 0 { - trc::event!( - Cluster(ClusterEvent::PeerHasChanges), - RemoteIp = peer.addr, - Details = "blocked_ips" - ); - - update_lists = true; - } - } - } - - continue 'outer; - } - } else if !remove_seeds { - remove_seeds = true; - } - } - - // Add new peer to the list. - trc::event!(Cluster(ClusterEvent::PeerDiscovered), RemoteIp = peer.addr); - self.peers.push(peer.into()); - } - - if remove_seeds { - self.peers.retain(|peer| !peer.is_seed()); - } - - if send_pong { - self.send_gossip(self.peers[0].addr, Request::Pong(self.build_peer_status())) - .await; - } - - // Reload settings - if update_config || update_lists { - let server = self.inner.build_server(); - - tokio::spawn(async move { - let result = if update_config { - server.reload().await - } else { - server.reload_blocked_ips().await - }; - match result { - Ok(result) => { - if let Some(new_core) = result.new_core { - // Update core - server.inner.shared_core.store(new_core.into()); - - // Reload ACME - if server - .inner - .ipc - .housekeeper_tx - .send(HousekeeperEvent::ReloadSettings) - .await - .is_err() - { - trc::event!( - Server(trc::ServerEvent::ThreadError), - Details = "Failed to send setting reload event to housekeeper", - CausedBy = trc::location!(), - ); - } - } - } - Err(err) => { - trc::error!( - err.details("Failed to reload settings") - .caused_by(trc::location!()) - ); - } - } - }); - } - } -} diff --git a/crates/services/src/gossip/request.rs b/crates/services/src/gossip/request.rs deleted file mode 100644 index c5c8e6e8..00000000 --- a/crates/services/src/gossip/request.rs +++ /dev/null @@ -1,114 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -use super::{EpochId, PeerStatus}; - -use common::auth::oauth::crypto::SymmetricEncrypt; -use std::net::IpAddr; -use utils::codec::leb128::Leb128_; - -#[derive(Debug)] -pub enum Request { - Ping(Vec), - Pong(Vec), - Leave(Vec), -} - -impl Request { - const PING: u8 = 0; - const PONG: u8 = 1; - const LEAVE: u8 = 2; - - pub fn from_bytes(bytes: &[u8]) -> Option { - let mut it = bytes.iter(); - let flags = it.next().copied()?; - let is_ipv6 = flags & (1 << 7) != 0; - - let mut peers = Vec::with_capacity(bytes.len() / std::mem::size_of::()); - 'outer: loop { - let addr = if !is_ipv6 { - let mut octets = [0u8; 4]; - for octet in octets.iter_mut() { - if let Some(byte) = it.next() { - *octet = *byte; - } else { - break 'outer; - } - } - IpAddr::V4(octets.into()) - } else { - let mut octets = [0u8; 16]; - for octet in octets.iter_mut() { - if let Some(byte) = it.next() { - *octet = *byte; - } else { - break 'outer; - } - } - IpAddr::V6(octets.into()) - }; - - peers.push(PeerStatus { - addr, - epoch: EpochId::from_leb128_it(&mut it)?, - gen_config: it.next().copied()?, - gen_lists: it.next().copied()?, - }); - } - match flags & !(1 << 7) { - 0 => Request::Ping(peers), - 1 => Request::Pong(peers), - 2 => Request::Leave(peers), - _ => return None, - } - .into() - } - - pub fn to_bytes(&self) -> Vec { - let (mut flag, peers) = match self { - Request::Ping(peers) => (Self::PING, peers), - Request::Pong(peers) => (Self::PONG, peers), - Request::Leave(peers) => (Self::LEAVE, peers), - }; - - debug_assert!(!peers.is_empty()); - - let mut bytes = Vec::with_capacity( - std::mem::size_of::() - + (peers.len() * std::mem::size_of::()) - + SymmetricEncrypt::ENCRYPT_TAG_LEN, - ); - - let is_ipv6 = peers.iter().any(|peer| peer.addr.is_ipv6()); - if is_ipv6 { - flag |= 1 << 7; - } - - bytes.push(flag); - - for peer in peers { - if !is_ipv6 { - match &peer.addr { - IpAddr::V4(addr) => bytes.extend_from_slice(addr.octets().as_slice()), - IpAddr::V6(_) => unreachable!(), - } - } else { - match &peer.addr { - IpAddr::V6(addr) => bytes.extend_from_slice(addr.octets().as_slice()), - IpAddr::V4(addr) => { - bytes.extend_from_slice(addr.to_ipv6_mapped().octets().as_slice()) - } - } - } - - peer.epoch.to_leb128_bytes(&mut bytes); - bytes.push(peer.gen_config); - bytes.push(peer.gen_lists); - } - - bytes - } -} diff --git a/crates/services/src/gossip/spawn.rs b/crates/services/src/gossip/spawn.rs deleted file mode 100644 index cbc2eac1..00000000 --- a/crates/services/src/gossip/spawn.rs +++ /dev/null @@ -1,223 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -use super::request::Request; -use super::{Gossiper, Peer, UDP_MAX_PAYLOAD}; -use common::auth::oauth::crypto::SymmetricEncrypt; -use common::{IPC_CHANNEL_BUFFER, Inner}; -use std::net::IpAddr; -use std::time::{Duration, Instant}; -use std::{net::SocketAddr, sync::Arc}; -use tokio::sync::watch; -use tokio::{net::UdpSocket, sync::mpsc}; -use utils::config::Config; - -pub struct GossiperBuilder { - bind_addr: IpAddr, - advertise_addr: IpAddr, - port: u16, - cluster_key: String, - peers: Vec, - ping_interval: Duration, -} - -// Quidnunc: an inquisitive and gossipy person, from Latin quid nunc? 'what now?'. -struct Quidnunc { - socket: UdpSocket, - nonce: Vec, - encryptor: SymmetricEncrypt, -} - -impl GossiperBuilder { - pub fn try_parse(config: &mut Config) -> Option { - // Load configuration - let bind_addr = config.property::("cluster.bind-addr")?; - let mut builder = GossiperBuilder { - bind_addr, - cluster_key: config - .value("cluster.key") - .filter(|s| !s.is_empty())? - .to_string(), - advertise_addr: config - .property::("cluster.advertise-addr") - .unwrap_or(bind_addr), - port: config - .property_or_default::("cluster.bind-port", "1179") - .unwrap_or(1179), - ping_interval: config - .property_or_default("cluster.heartbeat", "1s") - .unwrap_or(Duration::from_secs(1)), - peers: Vec::new(), - }; - - for (_, addr) in config.properties::("cluster.seed-nodes") { - if addr != builder.bind_addr && addr != builder.advertise_addr { - builder.peers.push(Peer::new_seed(addr)); - } - } - - builder.into() - } - - pub async fn spawn(self, inner: Arc, mut shutdown_rx: watch::Receiver) { - // Bind port - let quidnunc = Arc::new(Quidnunc { - socket: match UdpSocket::bind(SocketAddr::new(self.bind_addr, self.port)).await { - Ok(socket) => socket, - Err(e) => { - trc::event!( - Network(trc::NetworkEvent::BindError), - Details = "Failed to bind UDP socket", - LocalIp = self.bind_addr, - Reason = e.to_string() - ); - return; - } - }, - nonce: b"428934328968".to_vec(), - encryptor: SymmetricEncrypt::new( - self.cluster_key.as_bytes(), - "gossipmonger context key", - ), - }); - - trc::event!( - Network(trc::NetworkEvent::ListenStart), - LocalIp = self.bind_addr, - LocalPort = self.port, - ); - - // Create gossiper - let (gossip_tx, mut gossip_rx) = mpsc::channel::<(SocketAddr, Request)>(IPC_CHANNEL_BUFFER); - let mut gossiper = Gossiper { - addr: self.advertise_addr, - port: self.port, - epoch: 0, - peers: self.peers, - last_peer_pinged: u32::MAX as usize, - inner, - gossip_tx, - }; - let quidnunc_ = quidnunc.clone(); - let bind_addr = self.bind_addr; - let bind_port = self.port; - - // Spawn gossip sender - tokio::spawn(async move { - while let Some((target_addr, response)) = gossip_rx.recv().await { - // Encrypt packets - let mut bytes = response.to_bytes(); - match quidnunc_ - .encryptor - .encrypt_in_place(&mut bytes, &quidnunc_.nonce) - { - Ok(_) => { - if let Err(err) = quidnunc_.socket.send_to(&bytes, &target_addr).await { - trc::event!( - Network(trc::NetworkEvent::WriteError), - RemoteIp = target_addr.ip(), - RemotePort = target_addr.port(), - LocalIp = bind_addr, - LocalPort = bind_port, - Reason = err.to_string() - ); - } - } - Err(err) => { - trc::event!( - Cluster(trc::ClusterEvent::Error), - RemoteIp = target_addr.ip(), - RemotePort = target_addr.port(), - Reason = err, - Details = "Failed to encrypt UDP packet" - ); - } - } - } - }); - - // Spawn gossip listener - let ping_interval = self.ping_interval; - tokio::spawn(async move { - let mut buf = vec![0; UDP_MAX_PAYLOAD]; - let mut last_ping = Instant::now(); - let mut wait = ping_interval; - - loop { - tokio::select! { - packet = quidnunc.socket.recv_from(&mut buf) => { - match packet { - Ok((size, addr)) => { - // Decrypt packet - match quidnunc.encryptor.decrypt(&buf[..size], &quidnunc.nonce) { - Ok(bytes) => { - if let Some(request) = Request::from_bytes(&bytes) { - match request { - Request::Ping(peers) => { - gossiper.handle_ping(peers, true).await; - }, - Request::Pong(peers) => { - gossiper.handle_ping(peers, false).await; - }, - Request::Leave(peers) => { - gossiper.handle_leave(peers).await; - }, - } - } else { - trc::event!( - Cluster(trc::ClusterEvent::InvalidPacket), - RemoteIp = addr.ip(), - RemotePort = addr.port(), - Contents = bytes, - ); - } - }, - Err(err) => { - trc::event!( - Cluster(trc::ClusterEvent::DecryptionError), - RemoteIp = addr.ip(), - RemotePort = addr.port(), - Contents = (buf[..size]).to_vec(), - Reason = err, - ); - }, - } - } - Err(err) => { - trc::event!( - Network(trc::NetworkEvent::ReadError), - LocalIp = bind_addr, - LocalPort = bind_port, - Reason = err.to_string() - ); - } - } - }, - _ = tokio::time::sleep(wait) => { - // Send ping - gossiper.ping_peers().await; - last_ping = Instant::now(); - }, - _ = shutdown_rx.changed() => { - trc::event!( - Network(trc::NetworkEvent::ListenStop), - LocalIp = bind_addr, - LocalPort = bind_port, - ); - - // Broadcast leave message - gossiper.broadcast_leave().await; - - break; - } - }; - - // Calculate next ping interval - wait = ping_interval.saturating_sub(last_ping.elapsed()); - } - }); - } -} diff --git a/crates/services/src/housekeeper/mod.rs b/crates/services/src/housekeeper/mod.rs index d15f9645..69e255f6 100644 --- a/crates/services/src/housekeeper/mod.rs +++ b/crates/services/src/housekeeper/mod.rs @@ -15,7 +15,7 @@ use common::{ Inner, KV_LOCK_HOUSEKEEPER, LONG_1D_SLUMBER, Server, config::telemetry::OtelMetrics, core::BuildServer, - ipc::{HousekeeperEvent, PurgeType}, + ipc::{BroadcastEvent, HousekeeperEvent, PurgeType}, }; #[cfg(feature = "enterprise")] @@ -288,7 +288,9 @@ pub fn spawn_housekeeper(inner: Arc, mut rx: mpsc::Receiver, mut rx: mpsc::Receiver { diff --git a/crates/services/src/lib.rs b/crates/services/src/lib.rs index 5769b1f9..761f8fc7 100644 --- a/crates/services/src/lib.rs +++ b/crates/services/src/lib.rs @@ -4,8 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::sync::Arc; - +use broadcast::publisher::spawn_broadcast_publisher; use common::{ Inner, manager::boot::{BootManager, IpcReceivers}, @@ -13,8 +12,9 @@ use common::{ use housekeeper::spawn_housekeeper; use index::spawn_email_queue_task; use state_manager::manager::spawn_state_manager; +use std::sync::Arc; -pub mod gossip; +pub mod broadcast; pub mod housekeeper; pub mod index; pub mod state_manager; @@ -56,6 +56,12 @@ impl SpawnServices for IpcReceivers { // Spawn housekeeper spawn_housekeeper(inner.clone(), self.housekeeper_rx.take().unwrap()); + // Spawn broadcast publisher + if let Some(event_rx) = self.broadcast_rx.take() { + // Spawn broadcast publisher + spawn_broadcast_publisher(inner.clone(), event_rx); + } + // Spawn index task spawn_email_queue_task(inner); } diff --git a/crates/services/src/state_manager/http.rs b/crates/services/src/state_manager/http.rs index 26f72f41..edd01865 100644 --- a/crates/services/src/state_manager/http.rs +++ b/crates/services/src/state_manager/http.rs @@ -28,11 +28,11 @@ impl PushServer { tokio::spawn(async move { let mut response = StateChangeResponse::new(); for state_change in &state_changes { - for (type_state, change_id) in &state_change.types { + for type_state in state_change.types { response .changed .get_mut_or_insert(state_change.account_id.into()) - .set(*type_state, (*change_id).into()); + .set(type_state, (state_change.change_id).into()); } } diff --git a/crates/services/src/state_manager/manager.rs b/crates/services/src/state_manager/manager.rs index 86ebe04e..11706821 100644 --- a/crates/services/src/state_manager/manager.rs +++ b/crates/services/src/state_manager/manager.rs @@ -12,7 +12,7 @@ use std::{ use common::{ Inner, core::BuildServer, - ipc::{PushSubscription, StateEvent, UpdateSubscription}, + ipc::{BroadcastEvent, PushSubscription, StateEvent, UpdateSubscription}, }; use jmap_proto::types::{id::Id, state::StateChange, type_state::DataType}; use store::{ahash::AHashMap, rand}; @@ -135,7 +135,27 @@ pub fn spawn_state_manager(inner: Arc, mut change_rx: mpsc::Receiver { + StateEvent::Publish { + state_change, + broadcast, + } => { + // Publish event to cluster + if broadcast { + if let Some(broadcast_tx) = &inner.ipc.broadcast_tx.clone() { + if broadcast_tx + .send(BroadcastEvent::StateChange(state_change)) + .await + .is_err() + { + trc::event!( + Server(trc::ServerEvent::ThreadError), + Details = "Error sending broadcast event.", + CausedBy = trc::location!() + ); + } + } + } + if let Some(shared_accounts) = shared_accounts_map.get(&state_change.account_id) { let current_time = SystemTime::now() @@ -147,19 +167,18 @@ pub fn spawn_state_manager(inner: Arc, mut change_rx: mpsc::Receiver { let subscriber_tx = tx.clone(); - let state_change = state_change.clone(); tokio::spawn(async move { // Timeout after 500ms in case there is a blocked client @@ -167,6 +186,7 @@ pub fn spawn_state_manager(inner: Arc, mut change_rx: mpsc::Receiver) -> mpsc::Sender { Event::Push { ids, state_change } => { for id in ids { if let Some(subscription) = subscriptions.get_mut(&id) { - subscription.state_changes.push(state_change.clone()); + subscription.state_changes.push(state_change); let last_request = subscription.last_request.elapsed(); if !subscription.in_flight diff --git a/crates/store/Cargo.toml b/crates/store/Cargo.toml index 562efaa0..54cf6c41 100644 --- a/crates/store/Cargo.toml +++ b/crates/store/Cargo.toml @@ -12,6 +12,7 @@ rocksdb = { version = "0.23", optional = true, features = ["multi-threaded-cf"] foundationdb = { version = "0.9.2", features = ["embedded-fdb-include", "fdb-7_3"], optional = true } rusqlite = { version = "0.35", features = ["bundled"], optional = true } rust-s3 = { version = "0.35", default-features = false, features = ["tokio-rustls-tls", "no-verify-ssl"], optional = true } +async-nats = { version = "0.40", default-features = false, features = ["server_2_10", "server_2_11", "ring"], optional = true } azure_core = { version = "0.21.0", optional = true } azure_storage = { version = "0.21.0", default-features = false, features = ["enable_reqwest_rustls", "hmac_rust"], optional = true } azure_storage_blobs = { version = "0.21.0", default-features = false, features = ["enable_reqwest_rustls", "hmac_rust"], optional = true } @@ -44,7 +45,7 @@ serde_json = {version = "1.0.64", optional = true } regex = "1.7.0" flate2 = "1.0" async-trait = "0.1.68" -redis = { version = "0.29", features = [ "tokio-comp", "tokio-rustls-comp", "tls-rustls-insecure", "tls-rustls-webpki-roots", "cluster-async"], optional = true } +redis = { version = "0.31", features = [ "tokio-comp", "tokio-rustls-comp", "tls-rustls-insecure", "tls-rustls-webpki-roots", "cluster-async"], optional = true } deadpool = { version = "0.12", features = ["managed"], optional = true } arc-swap = "1.6.0" bitpacking = "0.9.2" @@ -62,6 +63,7 @@ sqlite = ["rusqlite", "rayon", "r2d2", "num_cpus", "lru-cache"] postgres = ["tokio-postgres", "deadpool-postgres", "tokio-rustls", "rustls", "ring", "rustls-pki-types", "futures", "bytes"] elastic = ["elasticsearch", "serde_json"] mysql = ["mysql_async", "futures"] +nats = ["async-nats"] s3 = ["rust-s3"] azure = ["azure_core", "azure_storage", "azure_storage_blobs"] foundation = ["foundationdb", "futures"] diff --git a/crates/store/src/backend/mod.rs b/crates/store/src/backend/mod.rs index 909021dc..62a63340 100644 --- a/crates/store/src/backend/mod.rs +++ b/crates/store/src/backend/mod.rs @@ -17,6 +17,8 @@ pub mod http; pub mod memory; #[cfg(feature = "mysql")] pub mod mysql; +#[cfg(feature = "nats")] +pub mod nats; #[cfg(feature = "postgres")] pub mod postgres; #[cfg(feature = "redis")] diff --git a/crates/store/src/backend/nats/mod.rs b/crates/store/src/backend/nats/mod.rs new file mode 100644 index 00000000..6df12940 --- /dev/null +++ b/crates/store/src/backend/nats/mod.rs @@ -0,0 +1,108 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use std::time::Duration; + +use async_nats::Client; +use utils::config::{Config, utils::AsKey}; + +pub mod pubsub; + +#[derive(Debug)] +pub struct NatsStore { + client: Client, +} + +impl NatsStore { + pub async fn open(config: &mut Config, prefix: impl AsKey) -> Option { + let prefix = prefix.as_key(); + let urls = config + .values((&prefix, "urls")) + .map(|(_, v)| v.to_string()) + .collect::>(); + if urls.is_empty() { + config.new_build_error((&prefix, "urls"), "No Nats URLs specified"); + return None; + } + + let mut opts = async_nats::ConnectOptions::new() + .max_reconnects( + config + .property_or_default::>((&prefix, "max-reconnects"), "false") + .unwrap_or_default(), + ) + .connection_timeout( + config + .property_or_default((&prefix, "timeout.connection"), "5s") + .unwrap_or_else(|| Duration::from_secs(5)), + ) + .request_timeout( + config + .property_or_default::>((&prefix, "timeout.request"), "10s") + .unwrap_or_else(|| Some(Duration::from_secs(10))), + ) + .ping_interval( + config + .property_or_default((&prefix, "ping-interval"), "60s") + .unwrap_or_else(|| Duration::from_secs(5)), + ) + .client_capacity( + config + .property_or_default((&prefix, "capacity.client"), "2048") + .unwrap_or(2048), + ) + .subscription_capacity( + config + .property_or_default((&prefix, "capacity.subscription"), "65536") + .unwrap_or(65536), + ) + .read_buffer_capacity( + config + .property_or_default((&prefix, "capacity.read-buffer"), "65535") + .unwrap_or(65535), + ) + .require_tls( + config + .property_or_default((&prefix, "tls.require"), "false") + .unwrap_or_default(), + ); + + if config + .property_or_default((&prefix, "no-echo"), "true") + .unwrap_or(true) + { + opts = opts.no_echo(); + } + + if let (Some(user), Some(pass)) = ( + config.value((&prefix, "user")), + config.value((&prefix, "password")), + ) { + opts = opts.user_and_password(user.to_string(), pass.to_string()); + } else if let Some(credentials) = config.value((&prefix, "credentials")) { + opts = opts + .credentials(credentials) + .map_err(|err| { + config.new_build_error( + (&prefix, "credentials"), + format!("Failed to parse Nats credentials: {}", err), + ); + }) + .ok()?; + } + + async_nats::connect_with_options(urls, opts) + .await + .map_err(|err| { + config.new_build_error( + (&prefix, "urls"), + format!("Failed to connect to Nats: {}", err), + ); + }) + .map(|client| NatsStore { client }) + .ok() + } +} diff --git a/crates/store/src/backend/nats/pubsub.rs b/crates/store/src/backend/nats/pubsub.rs new file mode 100644 index 00000000..81329a9e --- /dev/null +++ b/crates/store/src/backend/nats/pubsub.rs @@ -0,0 +1,45 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use std::fmt::Display; + +use futures::StreamExt; + +use crate::dispatch::pubsub::{Msg, PubSubStream}; + +use super::NatsStore; + +pub struct NatsPubSubStream { + subs: async_nats::Subscriber, +} + +impl NatsStore { + pub async fn publish(&self, topic: &'static str, message: Vec) -> trc::Result<()> { + self.client + .publish(topic, message.into()) + .await + .map_err(into_error) + } + + pub async fn subscribe(&self, topic: &'static str) -> trc::Result { + self.client + .subscribe(topic) + .await + .map_err(into_error) + .map(|subs| PubSubStream::Nats(NatsPubSubStream { subs })) + } +} + +impl NatsPubSubStream { + pub async fn next(&mut self) -> Option { + self.subs.next().await.map(Msg::Nats) + } +} + +#[inline(always)] +fn into_error(err: impl Display) -> trc::Error { + trc::StoreEvent::NatsError.reason(err) +} diff --git a/crates/store/src/backend/redis/mod.rs b/crates/store/src/backend/redis/mod.rs index 4faca371..ec48a480 100644 --- a/crates/store/src/backend/redis/mod.rs +++ b/crates/store/src/backend/redis/mod.rs @@ -11,13 +11,14 @@ use deadpool::{ managed::{Manager, Pool}, }; use redis::{ - Client, + Client, ProtocolVersion, cluster::{ClusterClient, ClusterClientBuilder}, }; use utils::config::{Config, utils::AsKey}; pub mod lookup; pub mod pool; +pub mod pubsub; #[derive(Debug)] pub struct RedisStore { @@ -105,6 +106,13 @@ impl RedisStore { if let Some(true) = config.property::((&prefix, "read-from-replicas")) { builder = builder.read_from_replicas(); } + if config + .value((&prefix, "protocol-version")) + .unwrap_or("resp2") + == "resp3" + { + builder = builder.use_protocol(ProtocolVersion::RESP3); + } let client = builder .build() diff --git a/crates/store/src/backend/redis/pubsub.rs b/crates/store/src/backend/redis/pubsub.rs new file mode 100644 index 00000000..228bc1ad --- /dev/null +++ b/crates/store/src/backend/redis/pubsub.rs @@ -0,0 +1,94 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use super::{RedisPool, RedisStore, into_error}; +use crate::dispatch::pubsub::{Msg, PubSubStream}; +use futures::StreamExt; +use redis::{AsyncCommands, PushInfo, cluster::ClusterConfig, cluster_async::ClusterConnection}; +use tokio::sync::mpsc::UnboundedReceiver; + +pub struct RedisPubSubStream { + stream: redis::aio::PubSubStream, +} + +pub struct RedisClusterPubSubStream { + _conn: ClusterConnection, + rx: UnboundedReceiver, +} + +impl RedisStore { + pub async fn publish(&self, topic: &'static str, message: Vec) -> trc::Result<()> { + match &self.pool { + RedisPool::Single(pool) => pool + .get() + .await + .map_err(into_error)? + .as_mut() + .publish(topic, message) + .await + .map_err(into_error), + RedisPool::Cluster(pool) => pool + .get() + .await + .map_err(into_error)? + .as_mut() + .publish(topic, message) + .await + .map_err(into_error), + } + } + + pub async fn subscribe(&self, topic: &'static str) -> trc::Result { + match &self.pool { + RedisPool::Single(pool) => { + let mut pubsub = pool + .manager() + .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(), + })) + } + RedisPool::Cluster(pool) => { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + + let mut _conn = pool + .manager() + .client + .get_async_connection_with_config(ClusterConfig::default().set_push_sender(tx)) + .await + .map_err(into_error)?; + + _conn.subscribe(topic).await.map_err(into_error)?; + + Ok(PubSubStream::RedisCluster(RedisClusterPubSubStream { + _conn, + rx, + })) + } + } + } +} + +impl RedisPubSubStream { + pub async fn next(&mut self) -> Option { + self.stream.next().await.map(Msg::Redis) + } +} + +impl RedisClusterPubSubStream { + pub async fn next(&mut self) -> Option { + loop { + if let Some(msg) = redis::Msg::from_push_info(self.rx.recv().await?) { + return Some(Msg::Redis(msg)); + } + } + } +} diff --git a/crates/store/src/config.rs b/crates/store/src/config.rs index 87b00915..717ac139 100644 --- a/crates/store/src/config.rs +++ b/crates/store/src/config.rs @@ -4,40 +4,15 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use std::sync::Arc; + use utils::config::{Config, cron::SimpleCron, utils::ParseValue}; use crate::{ - BlobStore, CompressionAlgo, InMemoryStore, PurgeSchedule, PurgeStore, Store, Stores, - backend::fs::FsStore, + BlobStore, CompressionAlgo, InMemoryStore, PubSubStore, PurgeSchedule, PurgeStore, Store, + Stores, backend::fs::FsStore, }; -#[cfg(feature = "s3")] -use crate::backend::s3::S3Store; - -#[cfg(feature = "postgres")] -use crate::backend::postgres::PostgresStore; - -#[cfg(feature = "mysql")] -use crate::backend::mysql::MysqlStore; - -#[cfg(feature = "sqlite")] -use crate::backend::sqlite::SqliteStore; - -#[cfg(feature = "foundation")] -use crate::backend::foundationdb::FdbStore; - -#[cfg(feature = "rocks")] -use crate::backend::rocksdb::RocksDbStore; - -#[cfg(feature = "elastic")] -use crate::backend::elastic::ElasticSearchStore; - -#[cfg(feature = "redis")] -use crate::backend::redis::RedisStore; - -#[cfg(feature = "azure")] -use crate::backend::azure::AzureStore; - #[cfg(feature = "enterprise")] enum CompositeStore { #[cfg(any(feature = "postgres", feature = "mysql"))] @@ -103,7 +78,10 @@ impl Stores { continue; } - if let Some(db) = RocksDbStore::open(config, prefix).await.map(Store::from) { + if let Some(db) = crate::backend::rocksdb::RocksDbStore::open(config, prefix) + .await + .map(Store::from) + { self.stores.insert(store_id.clone(), db.clone()); self.fts_stores.insert(store_id.clone(), db.clone().into()); self.blob_stores.insert( @@ -125,7 +103,10 @@ impl Stores { continue; } - if let Some(db) = FdbStore::open(config, prefix).await.map(Store::from) { + if let Some(db) = crate::backend::foundationdb::FdbStore::open(config, prefix) + .await + .map(Store::from) + { self.stores.insert(store_id.clone(), db.clone()); self.fts_stores.insert(store_id.clone(), db.clone().into()); self.blob_stores.insert( @@ -137,10 +118,13 @@ impl Stores { } #[cfg(feature = "postgres")] "postgresql" => { - if let Some(db) = - PostgresStore::open(config, prefix, config.is_active_store(id)) - .await - .map(Store::from) + if let Some(db) = crate::backend::postgres::PostgresStore::open( + config, + prefix, + config.is_active_store(id), + ) + .await + .map(Store::from) { self.stores.insert(store_id.clone(), db.clone()); self.fts_stores.insert(store_id.clone(), db.clone().into()); @@ -153,9 +137,13 @@ impl Stores { } #[cfg(feature = "mysql")] "mysql" => { - if let Some(db) = MysqlStore::open(config, prefix, config.is_active_store(id)) - .await - .map(Store::from) + if let Some(db) = crate::backend::mysql::MysqlStore::open( + config, + prefix, + config.is_active_store(id), + ) + .await + .map(Store::from) { self.stores.insert(store_id.clone(), db.clone()); self.fts_stores.insert(store_id.clone(), db.clone().into()); @@ -178,7 +166,9 @@ impl Stores { continue; } - if let Some(db) = SqliteStore::open(config, prefix).map(Store::from) { + if let Some(db) = + crate::backend::sqlite::SqliteStore::open(config, prefix).map(Store::from) + { self.stores.insert(store_id.clone(), db.clone()); self.fts_stores.insert(store_id.clone(), db.clone().into()); self.blob_stores.insert( @@ -196,27 +186,42 @@ impl Stores { } #[cfg(feature = "s3")] "s3" => { - if let Some(db) = S3Store::open(config, prefix).await.map(BlobStore::from) { + if let Some(db) = crate::backend::s3::S3Store::open(config, prefix) + .await + .map(BlobStore::from) + { self.blob_stores .insert(store_id, db.with_compression(compression_algo)); } } #[cfg(feature = "elastic")] "elasticsearch" => { - if let Some(db) = ElasticSearchStore::open(config, prefix) - .await - .map(crate::FtsStore::from) + if let Some(db) = + crate::backend::elastic::ElasticSearchStore::open(config, prefix) + .await + .map(crate::FtsStore::from) { self.fts_stores.insert(store_id, db); } } #[cfg(feature = "redis")] "redis" => { - if let Some(db) = RedisStore::open(config, prefix) + if let Some(db) = crate::backend::redis::RedisStore::open(config, prefix) .await - .map(InMemoryStore::from) + .map(Arc::new) { - self.in_memory_stores.insert(store_id, db); + self.in_memory_stores + .insert(store_id.clone(), InMemoryStore::Redis(db.clone())); + self.pubsub_stores.insert(store_id, PubSubStore::Redis(db)); + } + } + #[cfg(feature = "nats")] + "nats" => { + if let Some(db) = crate::backend::nats::NatsStore::open(config, prefix) + .await + .map(Arc::new) + { + self.pubsub_stores.insert(store_id, PubSubStore::Nats(db)); } } #[cfg(feature = "enterprise")] @@ -234,7 +239,10 @@ impl Stores { } #[cfg(feature = "azure")] "azure" => { - if let Some(db) = AzureStore::open(config, prefix).await.map(BlobStore::from) { + if let Some(db) = crate::backend::azure::AzureStore::open(config, prefix) + .await + .map(BlobStore::from) + { self.blob_stores .insert(store_id, db.with_compression(compression_algo)); } diff --git a/crates/store/src/dispatch/mod.rs b/crates/store/src/dispatch/mod.rs index b6d1c32a..0d919150 100644 --- a/crates/store/src/dispatch/mod.rs +++ b/crates/store/src/dispatch/mod.rs @@ -11,6 +11,7 @@ use crate::Store; pub mod blob; pub mod fts; pub mod lookup; +pub mod pubsub; pub mod store; impl Store { diff --git a/crates/store/src/dispatch/pubsub.rs b/crates/store/src/dispatch/pubsub.rs new file mode 100644 index 00000000..11fc1e64 --- /dev/null +++ b/crates/store/src/dispatch/pubsub.rs @@ -0,0 +1,93 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::PubSubStore; + +pub enum PubSubStream { + #[cfg(feature = "redis")] + Redis(crate::backend::redis::pubsub::RedisPubSubStream), + #[cfg(feature = "redis")] + RedisCluster(crate::backend::redis::pubsub::RedisClusterPubSubStream), + #[cfg(feature = "nats")] + Nats(crate::backend::nats::pubsub::NatsPubSubStream), + #[cfg(not(any(feature = "redis", feature = "nats")))] + Unimplemented, +} + +pub enum Msg { + #[cfg(feature = "redis")] + Redis(redis::Msg), + #[cfg(feature = "nats")] + Nats(async_nats::Message), + #[cfg(not(any(feature = "redis", feature = "nats")))] + Unimplemented, +} + +#[allow(unused_variables)] +impl PubSubStore { + pub async fn publish(&self, topic: &'static str, message: Vec) -> trc::Result<()> { + match self { + #[cfg(feature = "redis")] + PubSubStore::Redis(store) => store.publish(topic, message).await, + #[cfg(feature = "nats")] + PubSubStore::Nats(store) => store.publish(topic, message).await, + PubSubStore::None => Err(trc::StoreEvent::NotSupported.into_err()), + } + } + + pub async fn subscribe(&self, topic: &'static str) -> trc::Result { + match self { + #[cfg(feature = "redis")] + PubSubStore::Redis(store) => store.subscribe(topic).await, + #[cfg(feature = "nats")] + PubSubStore::Nats(store) => store.subscribe(topic).await, + PubSubStore::None => Err(trc::StoreEvent::NotSupported.into_err()), + } + } + + pub fn is_none(&self) -> bool { + matches!(self, PubSubStore::None) + } +} + +impl PubSubStream { + pub async fn next(&mut self) -> Option { + match self { + #[cfg(feature = "redis")] + PubSubStream::Redis(stream) => stream.next().await, + #[cfg(feature = "redis")] + PubSubStream::RedisCluster(stream) => stream.next().await, + #[cfg(feature = "nats")] + PubSubStream::Nats(stream) => stream.next().await, + #[cfg(not(any(feature = "redis", feature = "nats")))] + PubSubStream::Unimplemented => None, + } + } +} + +impl Msg { + pub fn payload(&self) -> &[u8] { + match self { + #[cfg(feature = "redis")] + Msg::Redis(msg) => msg.get_payload_bytes(), + #[cfg(feature = "nats")] + Msg::Nats(msg) => msg.payload.as_ref(), + #[cfg(not(any(feature = "redis", feature = "nats")))] + Msg::Unimplemented => &[], + } + } + + pub fn topic(&self) -> &str { + match self { + #[cfg(feature = "redis")] + Msg::Redis(msg) => msg.get_channel_name(), + #[cfg(feature = "nats")] + Msg::Nats(msg) => msg.subject.as_str(), + #[cfg(not(any(feature = "redis", feature = "nats")))] + Msg::Unimplemented => "", + } + } +} diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index f38a99f2..5c635074 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -28,33 +28,6 @@ use backend::{fs::FsStore, http::HttpStore, memory::StaticMemoryStore}; use utils::config::cron::SimpleCron; use write::{BitmapClass, ValueClass}; -#[cfg(feature = "s3")] -use backend::s3::S3Store; - -#[cfg(feature = "postgres")] -use backend::postgres::PostgresStore; - -#[cfg(feature = "mysql")] -use backend::mysql::MysqlStore; - -#[cfg(feature = "sqlite")] -use backend::sqlite::SqliteStore; - -#[cfg(feature = "foundation")] -use backend::foundationdb::FdbStore; - -#[cfg(feature = "rocks")] -use backend::rocksdb::RocksDbStore; - -#[cfg(feature = "elastic")] -use backend::elastic::ElasticSearchStore; - -#[cfg(feature = "redis")] -use backend::redis::RedisStore; - -#[cfg(feature = "azure")] -use backend::azure::AzureStore; - pub trait Deserialize: Sized + Sync + Send { fn deserialize(bytes: &[u8]) -> trc::Result; fn deserialize_owned(bytes: Vec) -> trc::Result { @@ -210,21 +183,22 @@ pub struct Stores { pub blob_stores: AHashMap, pub fts_stores: AHashMap, pub in_memory_stores: AHashMap, + pub pubsub_stores: AHashMap, pub purge_schedules: Vec, } #[derive(Clone, Default)] pub enum Store { #[cfg(feature = "sqlite")] - SQLite(Arc), + SQLite(Arc), #[cfg(feature = "foundation")] - FoundationDb(Arc), + FoundationDb(Arc), #[cfg(feature = "postgres")] - PostgreSQL(Arc), + PostgreSQL(Arc), #[cfg(feature = "mysql")] - MySQL(Arc), + MySQL(Arc), #[cfg(feature = "rocks")] - RocksDb(Arc), + RocksDb(Arc), #[cfg(all(feature = "enterprise", any(feature = "postgres", feature = "mysql")))] SQLReadReplica(Arc), #[default] @@ -248,9 +222,9 @@ pub enum BlobBackend { Store(Store), Fs(Arc), #[cfg(feature = "s3")] - S3(Arc), + S3(Arc), #[cfg(feature = "azure")] - Azure(Arc), + Azure(Arc), #[cfg(feature = "enterprise")] Sharded(Arc), } @@ -259,51 +233,61 @@ pub enum BlobBackend { pub enum FtsStore { Store(Store), #[cfg(feature = "elastic")] - ElasticSearch(Arc), + ElasticSearch(Arc), } #[derive(Clone, Debug)] pub enum InMemoryStore { Store(Store), #[cfg(feature = "redis")] - Redis(Arc), + Redis(Arc), Http(Arc), Static(Arc), #[cfg(feature = "enterprise")] Sharded(Arc), } +#[derive(Clone, Default)] +pub enum PubSubStore { + #[cfg(feature = "redis")] + Redis(Arc), + #[cfg(feature = "nats")] + Nats(Arc), + #[default] + None, +} + #[cfg(feature = "sqlite")] -impl From for Store { - fn from(store: SqliteStore) -> Self { +impl From for Store { + fn from(store: backend::sqlite::SqliteStore) -> Self { Self::SQLite(Arc::new(store)) } } #[cfg(feature = "foundation")] -impl From for Store { - fn from(store: FdbStore) -> Self { +impl From for Store { + fn from(store: backend::foundationdb::FdbStore) -> Self { Self::FoundationDb(Arc::new(store)) } } #[cfg(feature = "postgres")] -impl From for Store { - fn from(store: PostgresStore) -> Self { +impl From for Store { + fn from(store: backend::postgres::PostgresStore) -> Self { Self::PostgreSQL(Arc::new(store)) } } #[cfg(feature = "mysql")] -impl From for Store { - fn from(store: MysqlStore) -> Self { +impl From for Store { + fn from(store: backend::mysql::MysqlStore) -> Self { Self::MySQL(Arc::new(store)) } } #[cfg(feature = "rocks")] -impl From for Store { - fn from(store: RocksDbStore) -> Self { +impl From for Store { + fn from(store: backend::rocksdb::RocksDbStore) -> Self { Self::RocksDb(Arc::new(store)) } } @@ -318,8 +302,8 @@ impl From for BlobStore { } #[cfg(feature = "s3")] -impl From for BlobStore { - fn from(store: S3Store) -> Self { +impl From for BlobStore { + fn from(store: backend::s3::S3Store) -> Self { BlobStore { backend: BlobBackend::S3(Arc::new(store)), compression: CompressionAlgo::None, @@ -328,8 +312,8 @@ impl From for BlobStore { } #[cfg(feature = "azure")] -impl From for BlobStore { - fn from(store: AzureStore) -> Self { +impl From for BlobStore { + fn from(store: backend::azure::AzureStore) -> Self { BlobStore { backend: BlobBackend::Azure(Arc::new(store)), compression: CompressionAlgo::None, @@ -338,15 +322,15 @@ impl From for BlobStore { } #[cfg(feature = "elastic")] -impl From for FtsStore { - fn from(store: ElasticSearchStore) -> Self { +impl From for FtsStore { + fn from(store: backend::elastic::ElasticSearchStore) -> Self { Self::ElasticSearch(Arc::new(store)) } } #[cfg(feature = "redis")] -impl From for InMemoryStore { - fn from(store: RedisStore) -> Self { +impl From for InMemoryStore { + fn from(store: backend::redis::RedisStore) -> Self { Self::Redis(Arc::new(store)) } } diff --git a/crates/store/src/write/batch.rs b/crates/store/src/write/batch.rs index 432ac946..73035899 100644 --- a/crates/store/src/write/batch.rs +++ b/crates/store/src/write/batch.rs @@ -9,18 +9,11 @@ use super::{ ValueClass, ValueOp, assert::ToAssertValue, }; use crate::{SerializeInfallible, U32_LEN}; -use std::sync::{ - LazyLock, - atomic::{AtomicU64, Ordering}, -}; use utils::{ map::{bitmap::ShortId, vec_map::VecMap}, - snowflake::SnowflakeIdGenerator, + snowflake::HlcTimestamp, }; -static CHANGE_SEQ: AtomicU64 = AtomicU64::new(0); -static NODE_MUM: LazyLock = LazyLock::new(|| CHANGE_SEQ.swap(0, Ordering::Relaxed) as u16); - impl BatchBuilder { pub fn new() -> Self { Self { @@ -38,15 +31,8 @@ impl BatchBuilder { } } - pub fn init_id_generator(node_number: u16) { - CHANGE_SEQ.store(node_number as u64, Ordering::Relaxed); - } - fn generate_change_id(&mut self) -> u64 { - let change_id = SnowflakeIdGenerator::from_params( - CHANGE_SEQ.fetch_add(1, Ordering::Relaxed), - *NODE_MUM, - ); + let change_id = HlcTimestamp::generate(); self.current_change_id = Some(change_id); change_id } diff --git a/crates/trc/src/event/description.rs b/crates/trc/src/event/description.rs index 27685060..c7e50966 100644 --- a/crates/trc/src/event/description.rs +++ b/crates/trc/src/event/description.rs @@ -137,37 +137,35 @@ impl HttpEvent { impl ClusterEvent { pub fn description(&self) -> &'static str { match self { - ClusterEvent::PeerAlive => "A peer is alive", - ClusterEvent::PeerDiscovered => "A new peer was discovered", - ClusterEvent::PeerOffline => "A peer went offline", - ClusterEvent::PeerSuspected => "A peer is suspected to be offline", - ClusterEvent::PeerSuspectedIsAlive => "A suspected peer is actually alive", - ClusterEvent::PeerBackOnline => "A peer came back online", - ClusterEvent::PeerLeaving => "A peer is leaving the cluster", - ClusterEvent::PeerHasChanges => "A peer has reported changes", - ClusterEvent::OneOrMorePeersOffline => "One or more peers are offline", - ClusterEvent::EmptyPacket => "Received an empty gossip packet", - ClusterEvent::InvalidPacket => "Received an invalid gossip packet", - ClusterEvent::DecryptionError => "Failed to decrypt a gossip packet", - ClusterEvent::Error => "A cluster error occurred", + ClusterEvent::SubscriberStart => "PubSub subscriber started", + ClusterEvent::SubscriberStop => "PubSub subscriber stopped", + ClusterEvent::SubscriberError => "PubSub subscriber error", + ClusterEvent::SubscriberDisconnected => "PubSub subscriber disconnected", + ClusterEvent::PublisherStart => "PubSub publisher started", + ClusterEvent::PublisherStop => "PubSub publisher stopped", + ClusterEvent::PublisherError => "PubSub publisher error", + ClusterEvent::MessageReceived => "PubSub message received", + ClusterEvent::MessageSkipped => "PubSub message skipped", + ClusterEvent::MessageInvalid => "Invalid PubSub message", + ClusterEvent::ClockSkewDetected => "Clock skew detected", } } pub fn explain(&self) -> &'static str { match self { - ClusterEvent::PeerAlive => "A peer is alive and reachable", - ClusterEvent::PeerDiscovered => "A new peer was discovered", - ClusterEvent::PeerOffline => "A peer is offline", - ClusterEvent::PeerSuspected => "A peer is suspected to be offline", - ClusterEvent::PeerSuspectedIsAlive => "A suspected peer is actually alive", - ClusterEvent::PeerBackOnline => "A peer came back online", - ClusterEvent::PeerLeaving => "A peer is leaving the cluster", - ClusterEvent::PeerHasChanges => "A peer has reported changes", - ClusterEvent::OneOrMorePeersOffline => "One or more peers are offline", - ClusterEvent::EmptyPacket => "Received an empty gossip packet", - ClusterEvent::InvalidPacket => "Received an invalid gossip packet", - ClusterEvent::DecryptionError => "Failed to decrypt a gossip packet", - ClusterEvent::Error => "An error occurred in the cluster", + ClusterEvent::SubscriberStart => "The PubSub subscriber has started", + ClusterEvent::SubscriberStop => "The PubSub subscriber has stopped", + ClusterEvent::SubscriberError => "An error occurred while subscribing to PubSub", + ClusterEvent::SubscriberDisconnected => "The PubSub subscriber has disconnected", + ClusterEvent::PublisherStart => "The PubSub publisher has started", + ClusterEvent::PublisherStop => "The PubSub publisher has stopped", + ClusterEvent::PublisherError => "An error occurred while publishing to PubSub", + ClusterEvent::MessageReceived => "A message was received from the PubSub server", + ClusterEvent::MessageSkipped => "A message originating from this node was skipped", + ClusterEvent::MessageInvalid => { + "An invalid message was received from the PubSub server" + } + ClusterEvent::ClockSkewDetected => "A large clock skew was detected between nodes", } } } @@ -1559,6 +1557,7 @@ impl StoreEvent { StoreEvent::DataIterate => "Data store iteration operation", StoreEvent::HttpStoreFetch => "HTTP store updated", StoreEvent::HttpStoreError => "Error updating HTTP store", + StoreEvent::NatsError => "NATS error", } } @@ -1596,6 +1595,7 @@ impl StoreEvent { StoreEvent::DataIterate => "A data store iteration operation was executed", StoreEvent::HttpStoreFetch => "The HTTP store was updated", StoreEvent::HttpStoreError => "An error occurred while updating the HTTP store", + StoreEvent::NatsError => "A NATS error occurred", } } } diff --git a/crates/trc/src/event/level.rs b/crates/trc/src/event/level.rs index 105fe9b4..d508d34b 100644 --- a/crates/trc/src/event/level.rs +++ b/crates/trc/src/event/level.rs @@ -30,6 +30,7 @@ impl EventType { | StoreEvent::LdapError | StoreEvent::ElasticsearchError | StoreEvent::RedisError + | StoreEvent::NatsError | StoreEvent::S3Error | StoreEvent::AzureError | StoreEvent::FilesystemError @@ -356,18 +357,17 @@ impl EventType { PushSubscriptionEvent::Success => Level::Trace, }, EventType::Cluster(event) => match event { - ClusterEvent::PeerAlive - | ClusterEvent::PeerDiscovered - | ClusterEvent::PeerOffline - | ClusterEvent::PeerSuspected - | ClusterEvent::PeerSuspectedIsAlive - | ClusterEvent::PeerBackOnline - | ClusterEvent::PeerLeaving => Level::Info, - ClusterEvent::PeerHasChanges | ClusterEvent::OneOrMorePeersOffline => Level::Debug, - ClusterEvent::EmptyPacket - | ClusterEvent::Error - | ClusterEvent::DecryptionError - | ClusterEvent::InvalidPacket => Level::Warn, + ClusterEvent::SubscriberStart + | ClusterEvent::SubscriberStop + | ClusterEvent::PublisherStart + | ClusterEvent::PublisherStop => Level::Info, + ClusterEvent::SubscriberDisconnected | ClusterEvent::ClockSkewDetected => { + Level::Warn + } + ClusterEvent::MessageReceived | ClusterEvent::MessageSkipped => Level::Trace, + ClusterEvent::PublisherError + | ClusterEvent::SubscriberError + | ClusterEvent::MessageInvalid => Level::Error, }, EventType::Housekeeper(event) => match event { HousekeeperEvent::Start | HousekeeperEvent::Stop => Level::Info, diff --git a/crates/trc/src/ipc/metrics.rs b/crates/trc/src/ipc/metrics.rs index a4929af8..698fa5c3 100644 --- a/crates/trc/src/ipc/metrics.rs +++ b/crates/trc/src/ipc/metrics.rs @@ -586,13 +586,9 @@ impl EventType { ) => true, EventType::PushSubscription(_) => true, EventType::Cluster( - ClusterEvent::PeerOffline - | ClusterEvent::PeerSuspected - | ClusterEvent::PeerSuspectedIsAlive - | ClusterEvent::EmptyPacket - | ClusterEvent::InvalidPacket - | ClusterEvent::DecryptionError - | ClusterEvent::Error, + ClusterEvent::SubscriberError + | ClusterEvent::PublisherError + | ClusterEvent::SubscriberDisconnected, ) => true, EventType::Housekeeper(_) => false, EventType::TaskQueue( diff --git a/crates/trc/src/lib.rs b/crates/trc/src/lib.rs index 773f1b01..2a67930f 100644 --- a/crates/trc/src/lib.rs +++ b/crates/trc/src/lib.rs @@ -214,19 +214,17 @@ pub enum SecurityEvent { #[event_type] pub enum ClusterEvent { - PeerAlive, - PeerDiscovered, - PeerOffline, - PeerSuspected, - PeerSuspectedIsAlive, - PeerBackOnline, - PeerLeaving, - PeerHasChanges, - OneOrMorePeersOffline, - EmptyPacket, - InvalidPacket, - DecryptionError, - Error, + SubscriberStart, + SubscriberStop, + SubscriberError, + SubscriberDisconnected, + PublisherStart, + PublisherStop, + PublisherError, + MessageReceived, + MessageSkipped, + MessageInvalid, + ClockSkewDetected, } #[event_type] @@ -830,6 +828,7 @@ pub enum StoreEvent { ElasticsearchError, RedisError, S3Error, + NatsError, AzureError, FilesystemError, PoolError, diff --git a/crates/trc/src/serializers/binary.rs b/crates/trc/src/serializers/binary.rs index d0e06ec5..8bb2e6af 100644 --- a/crates/trc/src/serializers/binary.rs +++ b/crates/trc/src/serializers/binary.rs @@ -348,19 +348,17 @@ impl EventType { EventType::Auth(AuthEvent::MissingTotp) => 36, EventType::Auth(AuthEvent::Success) => 37, EventType::Auth(AuthEvent::TooManyAttempts) => 38, - EventType::Cluster(ClusterEvent::DecryptionError) => 39, - EventType::Cluster(ClusterEvent::EmptyPacket) => 40, - EventType::Cluster(ClusterEvent::Error) => 41, - EventType::Cluster(ClusterEvent::InvalidPacket) => 42, - EventType::Cluster(ClusterEvent::OneOrMorePeersOffline) => 43, - EventType::Cluster(ClusterEvent::PeerAlive) => 44, - EventType::Cluster(ClusterEvent::PeerBackOnline) => 45, - EventType::Cluster(ClusterEvent::PeerDiscovered) => 46, - EventType::Cluster(ClusterEvent::PeerHasChanges) => 47, - EventType::Cluster(ClusterEvent::PeerLeaving) => 49, - EventType::Cluster(ClusterEvent::PeerOffline) => 50, - EventType::Cluster(ClusterEvent::PeerSuspected) => 51, - EventType::Cluster(ClusterEvent::PeerSuspectedIsAlive) => 52, + EventType::Cluster(ClusterEvent::SubscriberStart) => 39, + EventType::Cluster(ClusterEvent::SubscriberStop) => 40, + EventType::Cluster(ClusterEvent::SubscriberError) => 41, + EventType::Cluster(ClusterEvent::SubscriberDisconnected) => 42, + EventType::Cluster(ClusterEvent::PublisherStart) => 43, + EventType::Cluster(ClusterEvent::PublisherStop) => 44, + EventType::Cluster(ClusterEvent::PublisherError) => 45, + EventType::Cluster(ClusterEvent::MessageReceived) => 46, + EventType::Cluster(ClusterEvent::MessageSkipped) => 47, + EventType::Cluster(ClusterEvent::MessageInvalid) => 49, + EventType::Cluster(ClusterEvent::ClockSkewDetected) => 50, EventType::Config(ConfigEvent::AlreadyUpToDate) => 53, EventType::Config(ConfigEvent::BuildError) => 54, EventType::Config(ConfigEvent::BuildWarning) => 55, @@ -886,6 +884,7 @@ impl EventType { EventType::WebDav(WebDavEvent::Head) => 574, EventType::WebDav(WebDavEvent::Mkcalendar) => 575, EventType::Calendar(CalendarEvent::RuleExpansionError) => 576, + EventType::Store(StoreEvent::NatsError) => 577, } } @@ -930,19 +929,17 @@ impl EventType { 36 => Some(EventType::Auth(AuthEvent::MissingTotp)), 37 => Some(EventType::Auth(AuthEvent::Success)), 38 => Some(EventType::Auth(AuthEvent::TooManyAttempts)), - 39 => Some(EventType::Cluster(ClusterEvent::DecryptionError)), - 40 => Some(EventType::Cluster(ClusterEvent::EmptyPacket)), - 41 => Some(EventType::Cluster(ClusterEvent::Error)), - 42 => Some(EventType::Cluster(ClusterEvent::InvalidPacket)), - 43 => Some(EventType::Cluster(ClusterEvent::OneOrMorePeersOffline)), - 44 => Some(EventType::Cluster(ClusterEvent::PeerAlive)), - 45 => Some(EventType::Cluster(ClusterEvent::PeerBackOnline)), - 46 => Some(EventType::Cluster(ClusterEvent::PeerDiscovered)), - 47 => Some(EventType::Cluster(ClusterEvent::PeerHasChanges)), - 49 => Some(EventType::Cluster(ClusterEvent::PeerLeaving)), - 50 => Some(EventType::Cluster(ClusterEvent::PeerOffline)), - 51 => Some(EventType::Cluster(ClusterEvent::PeerSuspected)), - 52 => Some(EventType::Cluster(ClusterEvent::PeerSuspectedIsAlive)), + 39 => Some(EventType::Cluster(ClusterEvent::SubscriberStart)), + 40 => Some(EventType::Cluster(ClusterEvent::SubscriberStop)), + 41 => Some(EventType::Cluster(ClusterEvent::SubscriberError)), + 42 => Some(EventType::Cluster(ClusterEvent::SubscriberDisconnected)), + 43 => Some(EventType::Cluster(ClusterEvent::PublisherStart)), + 44 => Some(EventType::Cluster(ClusterEvent::PublisherStop)), + 45 => Some(EventType::Cluster(ClusterEvent::PublisherError)), + 46 => Some(EventType::Cluster(ClusterEvent::MessageReceived)), + 47 => Some(EventType::Cluster(ClusterEvent::MessageSkipped)), + 49 => Some(EventType::Cluster(ClusterEvent::MessageInvalid)), + 50 => Some(EventType::Cluster(ClusterEvent::ClockSkewDetected)), 53 => Some(EventType::Config(ConfigEvent::AlreadyUpToDate)), 54 => Some(EventType::Config(ConfigEvent::BuildError)), 55 => Some(EventType::Config(ConfigEvent::BuildWarning)), @@ -1508,11 +1505,14 @@ impl EventType { 574 => Some(EventType::WebDav(WebDavEvent::Head)), 575 => Some(EventType::WebDav(WebDavEvent::Mkcalendar)), 576 => Some(EventType::Calendar(CalendarEvent::RuleExpansionError)), + 577 => Some(EventType::Store(StoreEvent::NatsError)), _ => None, } } } +// 51, 52 + impl Key { fn code(&self) -> u64 { match self { diff --git a/crates/utils/src/snowflake.rs b/crates/utils/src/snowflake.rs index 4debe77c..3ea86e8c 100644 --- a/crates/utils/src/snowflake.rs +++ b/crates/utils/src/snowflake.rs @@ -5,17 +5,23 @@ */ use std::{ - sync::atomic::{AtomicU64, Ordering}, + sync::{ + LazyLock, + atomic::{AtomicU64, Ordering}, + }, time::{Duration, SystemTime}, }; #[derive(Debug)] pub struct SnowflakeIdGenerator { epoch: SystemTime, + last_timestamp: AtomicU64, node_id: u64, sequence: AtomicU64, } +pub struct HlcTimestamp; + const SEQUENCE_LEN: u64 = 12; const NODE_ID_LEN: u64 = 9; @@ -25,6 +31,12 @@ const NODE_ID_MASK: u64 = (1 << NODE_ID_LEN) - 1; const DEFAULT_EPOCH: u64 = 1632280000; // 52 years after UNIX_EPOCH const DEFAULT_EPOCH_MS: u128 = (DEFAULT_EPOCH as u128) * 1000; // 52 years after UNIX_EPOCH in milliseconds +const MAX_CLOCK_DRIFT: i64 = 60 * 1000; // 1 minute + +static LOGICAL_TIME: AtomicU64 = AtomicU64::new(0); +static CHANGE_SEQ: AtomicU64 = AtomicU64::new(0); +static NODE_MUM: LazyLock = LazyLock::new(|| CHANGE_SEQ.swap(0, Ordering::Relaxed) as u16); + /* ID characteristics: @@ -61,6 +73,7 @@ impl SnowflakeIdGenerator { epoch: SystemTime::UNIX_EPOCH + Duration::from_secs(DEFAULT_EPOCH), // 52 years after UNIX_EPOCH node_id, sequence: 0.into(), + last_timestamp: 0.into(), } } @@ -79,29 +92,68 @@ impl SnowflakeIdGenerator { #[inline(always)] pub fn generate(&self) -> u64 { - let elapsed = self + let current_elapsed = self .epoch .elapsed() .map(|e| e.as_millis()) .unwrap_or_default() as u64; - let sequence = self.sequence.fetch_add(1, Ordering::Relaxed); + let last_elapsed = self + .last_timestamp + .fetch_max(current_elapsed, Ordering::Relaxed); + let (elapsed, sequence) = if current_elapsed > last_elapsed { + (current_elapsed, 0) + } else { + (last_elapsed, self.sequence.fetch_add(1, Ordering::Relaxed)) + }; (elapsed << (SEQUENCE_LEN + NODE_ID_LEN)) - | ((self.node_id & NODE_ID_MASK) << SEQUENCE_LEN) - | (sequence & SEQUENCE_MASK) + | ((sequence & SEQUENCE_MASK) << NODE_ID_LEN) + | (self.node_id & NODE_ID_MASK) + } +} + +impl HlcTimestamp { + pub fn init(node_number: u16) { + CHANGE_SEQ.store(node_number as u64, Ordering::Relaxed); } - #[inline(always)] - pub fn from_params(sequence: u64, node_id: u16) -> u64 { - let elapsed = SystemTime::UNIX_EPOCH + pub fn update_clock_from_remote_timestamp(timestamp: u64) -> Result { + let remote_clock = timestamp >> (SEQUENCE_LEN + NODE_ID_LEN); + let local_elapsed = SystemTime::UNIX_EPOCH .elapsed() .map(|e| e.as_millis()) .unwrap_or_default() .saturating_sub(DEFAULT_EPOCH_MS) as u64; + let diff = remote_clock as i64 - local_elapsed as i64; + if diff > 0 { + if diff < MAX_CLOCK_DRIFT { + LOGICAL_TIME.fetch_max(remote_clock, Ordering::SeqCst); + Ok(diff) + } else { + Err(diff) + } + } else { + Ok(diff) + } + } + + pub fn generate() -> u64 { + let node_id = *NODE_MUM; + let current_elapsed = SystemTime::UNIX_EPOCH + .elapsed() + .map(|e| e.as_millis()) + .unwrap_or_default() + .saturating_sub(DEFAULT_EPOCH_MS) as u64; + let last_elapsed = LOGICAL_TIME.fetch_max(current_elapsed, Ordering::SeqCst); + let (elapsed, sequence) = if current_elapsed > last_elapsed { + (current_elapsed, 0) + } else { + (last_elapsed, CHANGE_SEQ.fetch_add(1, Ordering::Relaxed)) + }; (elapsed << (SEQUENCE_LEN + NODE_ID_LEN)) - | (((node_id as u64) & NODE_ID_MASK) << SEQUENCE_LEN) - | (sequence & SEQUENCE_MASK) + | ((sequence & SEQUENCE_MASK) << NODE_ID_LEN) + | (node_id as u64 & NODE_ID_MASK) } } @@ -116,6 +168,7 @@ impl Clone for SnowflakeIdGenerator { Self { epoch: self.epoch, node_id: self.node_id, + last_timestamp: 0.into(), sequence: 0.into(), } } diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 6245e429..ae7a0092 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -5,7 +5,7 @@ edition = "2024" resolver = "2" [features] -default = ["sqlite", "postgres", "mysql", "rocks", "elastic", "s3", "redis", "azure", "foundationdb"] +default = ["sqlite", "postgres", "mysql", "rocks", "elastic", "s3", "redis", "nats", "azure", "foundationdb"] #default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis"] #default = ["rocks", "redis", "s3"] sqlite = ["store/sqlite"] @@ -16,6 +16,7 @@ rocks = ["store/rocks"] elastic = ["store/elastic"] s3 = ["store/s3"] redis = ["store/redis"] +nats = ["store/nats"] azure = ["store/azure"] [dev-dependencies] diff --git a/tests/src/cluster/broadcast.rs b/tests/src/cluster/broadcast.rs new file mode 100644 index 00000000..c263c2d7 --- /dev/null +++ b/tests/src/cluster/broadcast.rs @@ -0,0 +1,31 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use std::net::IpAddr; + +use crate::imap::idle; + +use super::ClusterTest; + +pub async fn test(cluster: &ClusterTest) { + println!("Running cluster broadcast tests..."); + + // Run IMAP idle tests across nodes + let mut node1_client = cluster.imap_client("john", 1).await; + let mut node2_client = cluster.imap_client("john", 2).await; + idle::test(&mut node1_client, &mut node2_client, true).await; + + // Test event broadcast + let server1 = cluster.server(1); + let server2 = cluster.server(2); + let test_ip: IpAddr = "8.8.8.8".parse().unwrap(); + assert!(!server1.is_ip_blocked(&test_ip)); + assert!(!server2.is_ip_blocked(&test_ip)); + server1.block_ip(test_ip).await.unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!(server1.is_ip_blocked(&test_ip)); + assert!(server2.is_ip_blocked(&test_ip)); +} diff --git a/tests/src/cluster/mod.rs b/tests/src/cluster/mod.rs new file mode 100644 index 00000000..3f681964 --- /dev/null +++ b/tests/src/cluster/mod.rs @@ -0,0 +1,362 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use std::{path::PathBuf, sync::Arc, time::Duration}; + +use common::{ + Caches, Core, Data, Inner, Server, + config::{ + server::{Listeners, ServerProtocol}, + telemetry::Telemetry, + }, + core::BuildServer, + manager::{ + boot::build_ipc, + config::{ConfigManager, Patterns}, + }, +}; +use http::HttpSessionManager; +use imap::core::ImapSessionManager; +use imap_proto::ResponseType; +use jmap_client::client::{Client, Credentials}; +use managesieve::core::ManageSieveSessionManager; +use pop3::Pop3SessionManager; +use services::{SpawnServices, broadcast::subscriber::spawn_broadcast_subscriber}; +use smtp::{SpawnQueueManager, core::SmtpSessionManager}; +use store::Stores; +use tokio::sync::watch; +use utils::config::Config; + +use crate::{ + AssertConfig, TEST_USERS, add_test_certs, + directory::internal::TestInternalDirectory, + imap::{ImapConnection, Type}, + jmap::enterprise::EnterpriseCore, +}; + +pub mod broadcast; +pub mod stress; + +pub const NUM_NODES: usize = 3; + +#[tokio::test(flavor = "multi_thread")] +pub async fn cluster_tests() { + let params = init_cluster_tests(true).await; + //stress::test(params.server.clone(), params.client).await; + broadcast::test(¶ms).await; +} + +#[allow(dead_code)] +pub struct ClusterTest { + servers: Vec, + shutdown_txs: Vec>, +} + +async fn init_cluster_tests(delete_if_exists: bool) -> ClusterTest { + // Load and parse config + let store_id = std::env::var("STORE").expect( + "Missing store type. Try running `STORE= PUBSUB= cargo test`", + ); + let pubsub_id = std::env::var("PUBSUB").expect( + "Missing store type. Try running `STORE= PUBSUB= cargo test`", + ); + let mut pubsub_config = match pubsub_id.as_str() { + "nats" => Config::new(SERVER_NATS).unwrap(), + "redis" => Config::new(SERVER_REDIS).unwrap(), + _ => panic!("Unsupported pubsub type: {}", pubsub_id), + }; + + // Build configs + let mut configs = Vec::with_capacity(NUM_NODES); + for node_id in 0..NUM_NODES { + let mut config = Config::new( + add_test_certs(SERVER) + .replace("{STORE}", &store_id) + .replace("{PUBSUB}", &pubsub_id) + .replace("{NODE_ID}", &node_id.to_string()) + .replace( + "{LEVEL}", + &std::env::var("LOG").unwrap_or_else(|_| "disable".to_string()), + ), + ) + .unwrap(); + config.resolve_all_macros().await; + configs.push(config); + } + + // Build stores + let stores = Stores::parse_all(configs.first_mut().unwrap(), false).await; + + // Build servers + let mut servers = Vec::with_capacity(NUM_NODES); + let mut shutdown_txs = Vec::with_capacity(NUM_NODES); + for config in configs { + let mut stores = stores.clone(); + stores.pubsub_stores = Stores::parse(&mut pubsub_config).await.pubsub_stores; + let (server, shutdown_tx) = build_server(config, stores).await; + servers.push(server); + shutdown_txs.push(shutdown_tx); + } + + let store = servers.first().unwrap().store().clone(); + if delete_if_exists { + store.destroy().await; + } + + // Create test users + for (account, secret, name, email) in TEST_USERS { + let _account_id = store + .create_test_user(account, secret, name, &[email]) + .await; + } + + ClusterTest { + servers, + shutdown_txs, + } +} + +impl ClusterTest { + pub async fn jmap_client(&self, login: &str, node_id: u32) -> Client { + Client::new() + .credentials(Credentials::basic(login, find_account_secret(login))) + .timeout(Duration::from_secs(3600)) + .accept_invalid_certs(true) + .connect(&format!("https://127.0.0.1:1800{node_id}")) + .await + .unwrap() + } + + pub async fn imap_client(&self, login: &str, node_id: u32) -> ImapConnection { + let mut conn = ImapConnection::connect_to(b"A1 ", format!("127.0.0.1:1900{node_id}")).await; + conn.assert_read(Type::Untagged, ResponseType::Ok).await; + conn.authenticate(login, find_account_secret(login)).await; + conn + } + + pub fn server(&self, node_id: usize) -> &Server { + self.servers + .get(node_id) + .unwrap_or_else(|| panic!("No server found for node ID: {}", node_id)) + } +} + +fn find_account_secret(login: &str) -> &str { + TEST_USERS + .iter() + .find(|(account, _, _, _)| account == &login) + .map(|(_, secret, _, _)| secret) + .unwrap_or_else(|| panic!("No account found for login: {}", login)) +} + +async fn build_server(mut config: Config, stores: Stores) -> (Server, watch::Sender) { + // Parse servers + let mut servers = Listeners::parse(&mut config); + + // Bind ports and drop privileges + servers.bind_and_drop_priv(&mut config); + + // Parse core + let config_manager = ConfigManager { + cfg_local: Default::default(), + cfg_local_path: PathBuf::new(), + cfg_local_patterns: Patterns::parse(&mut config).into(), + cfg_store: config + .value("storage.data") + .and_then(|id| stores.stores.get(id)) + .cloned() + .unwrap_or_default(), + }; + let tracers = Telemetry::parse(&mut config, &stores); + let core = Core::parse(&mut config, stores, config_manager) + .await + .enable_enterprise(); + let data = Data::parse(&mut config); + let cache = Caches::parse(&mut config); + let (ipc, mut ipc_rxs) = build_ipc(&mut config, true); + let inner = Arc::new(Inner { + shared_core: core.into_shared(), + data, + ipc, + cache, + }); + + // Parse acceptors + servers.parse_tcp_acceptors(&mut config, inner.clone()); + + // Enable tracing + tracers.enable(true); + + // Start services + config.assert_no_errors(); + ipc_rxs.spawn_queue_manager(inner.clone()); + ipc_rxs.spawn_services(inner.clone()); + + // Spawn servers + let (shutdown_tx, shutdown_rx) = servers.spawn(|server, acceptor, shutdown_rx| { + match &server.protocol { + ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn( + SmtpSessionManager::new(inner.clone()), + inner.clone(), + acceptor, + shutdown_rx, + ), + ServerProtocol::Http => server.spawn( + HttpSessionManager::new(inner.clone()), + inner.clone(), + acceptor, + shutdown_rx, + ), + ServerProtocol::Imap => server.spawn( + ImapSessionManager::new(inner.clone()), + inner.clone(), + acceptor, + shutdown_rx, + ), + ServerProtocol::Pop3 => server.spawn( + Pop3SessionManager::new(inner.clone()), + inner.clone(), + acceptor, + shutdown_rx, + ), + ServerProtocol::ManageSieve => server.spawn( + ManageSieveSessionManager::new(inner.clone()), + inner.clone(), + acceptor, + shutdown_rx, + ), + }; + }); + + // Start broadcast subscriber + spawn_broadcast_subscriber(inner.clone(), shutdown_rx); + + (inner.build_server(), shutdown_tx) +} + +const SERVER: &str = r#" +[server] +hostname = "'server{NODE_ID}.example.org'" +http.url = "'https://127.0.0.1:800{NODE_ID}'" + +[cluster] +node-id = {NODE_ID} + +[server.listener.http] +bind = ["127.0.0.1:1800{NODE_ID}"] +protocol = "http" +max-connections = 81920 +tls.implicit = true + +[server.listener.imap] +bind = ["127.0.0.1:1900{NODE_ID}"] +protocol = "imap" +max-connections = 81920 + +[server.listener.lmtp] +bind = ['127.0.0.1:1700{NODE_ID}'] +protocol = 'lmtp' +tls.implicit = false + +[server.socket] +reuse-addr = true + +[server.tls] +enable = true +implicit = false +certificate = "default" + +[session.ehlo] +reject-non-fqdn = false + +[session.rcpt] +relay = [ { if = "!is_empty(authenticated_as)", then = true }, + { else = false } ] +directory = "'{STORE}'" + +[session.rcpt.errors] +total = 5 +wait = "1ms" + +[session.auth] +mechanisms = "[plain, login, oauthbearer]" +directory = "'{STORE}'" + +[resolver] +type = "system" + +[queue.outbound] +next-hop = [ { if = "rcpt_domain == 'example.com'", then = "'local'" }, + { if = "contains(['remote.org', 'foobar.com', 'test.com', 'other_domain.com'], rcpt_domain)", then = "'mock-smtp'" }, + { else = false } ] + +[store."foundationdb"] +type = "foundationdb" + +[store."postgresql"] +type = "postgresql" +host = "localhost" +port = 5432 +database = "stalwart" +user = "postgres" +password = "mysecretpassword" + +[store."mysql"] +type = "mysql" +host = "localhost" +port = 3307 +database = "stalwart" +user = "root" +password = "password" + +[certificate.default] +cert = "%{file:{CERT}}%" +private-key = "%{file:{PK}}%" + +[storage] +data = "{STORE}" +fts = "{STORE}" +blob = "{STORE}" +lookup = "{STORE}" +directory = "{STORE}" +pubsub = "{PUBSUB}" + +[directory."{STORE}"] +type = "internal" +store = "{STORE}" + +[imap.auth] +allow-plain-text = true + +[oauth] +key = "parerga_und_paralipomena" + +[spam-filter] +enable = false + +[tracer.console] +type = "console" +level = "{LEVEL}" +multiline = false +ansi = true +disabled-events = ["network.*", "telemetry.webhook-error", "http.request-body", + "eval.result", "store.*", "dkim.*", "queue.*", "delivery.*", + "*.raw-input", "*.raw-output" ] +"#; + +const SERVER_NATS: &str = r#" +[store."nats"] +type = "nats" +urls = "127.0.0.1:4444" +"#; + +const SERVER_REDIS: &str = r#" +[store."redis"] +type = "redis" +urls = "redis://127.0.0.1" +redis-type = "single" + +"#; diff --git a/tests/src/jmap/stress_test.rs b/tests/src/cluster/stress.rs similarity index 99% rename from tests/src/jmap/stress_test.rs rename to tests/src/cluster/stress.rs index 8c4dbabb..d1f50c25 100644 --- a/tests/src/jmap/stress_test.rs +++ b/tests/src/cluster/stress.rs @@ -4,10 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::{sync::Arc, time::Duration}; - -use super::assert_is_empty; -use crate::jmap::{mailbox::destroy_all_mailboxes_no_wait, wait_for_index}; +use crate::jmap::{assert_is_empty, mailbox::destroy_all_mailboxes_no_wait, wait_for_index}; use common::Server; use directory::backend::internal::manage::ManageDirectory; use email::{ @@ -21,6 +18,7 @@ use jmap_client::{ mailbox::{self, Mailbox, Role}, }; use jmap_proto::types::{collection::Collection, id::Id}; +use std::{sync::Arc, time::Duration}; use store::{ rand::{self, Rng}, roaring::RoaringBitmap, @@ -30,7 +28,7 @@ const TEST_USER_ID: u32 = 1; const NUM_PASSES: usize = 1; pub async fn test(server: Server, mut client: Client) { - println!("Running concurrency stress tests..."); + println!("Running cluster concurrency stress tests..."); server .core .storage diff --git a/tests/src/imap/idle.rs b/tests/src/imap/idle.rs index 47828561..d1991b55 100644 --- a/tests/src/imap/idle.rs +++ b/tests/src/imap/idle.rs @@ -4,13 +4,21 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use std::time::Duration; + use imap_proto::ResponseType; use crate::jmap::delivery::SmtpConnection; use super::{AssertResult, ImapConnection, Type}; -pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { +const SLEEP: Duration = Duration::from_millis(200); + +pub async fn test( + imap: &mut ImapConnection, + imap_check: &mut ImapConnection, + is_cluster_test: bool, +) { println!("Running IDLE tests..."); // Switch connection to IDLE mode @@ -28,6 +36,9 @@ pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { // Expect a new mailbox update imap.send("CREATE Provolone").await; imap.assert_read(Type::Tagged, ResponseType::Ok).await; + if is_cluster_test { + tokio::time::sleep(SLEEP).await; + } imap_check .assert_read(Type::Status, ResponseType::Ok) .await @@ -40,6 +51,9 @@ pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { imap.assert_read(Type::Continuation, ResponseType::Ok).await; imap.send_untagged(message).await; imap.assert_read(Type::Tagged, ResponseType::Ok).await; + if is_cluster_test { + tokio::time::sleep(SLEEP).await; + } imap_check .assert_read(Type::Status, ResponseType::Ok) .await @@ -53,6 +67,9 @@ pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { imap.assert_read(Type::Tagged, ResponseType::Ok).await; imap.send("STORE 1:* +FLAGS (\\Seen)").await; imap.assert_read(Type::Tagged, ResponseType::Ok).await; + if is_cluster_test { + tokio::time::sleep(SLEEP).await; + } imap_check .assert_read(Type::Status, ResponseType::Ok) .await @@ -66,6 +83,9 @@ pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { imap.assert_read(Type::Tagged, ResponseType::Ok).await; imap.send("CLOSE").await; imap.assert_read(Type::Tagged, ResponseType::Ok).await; + if is_cluster_test { + tokio::time::sleep(SLEEP).await; + } imap_check .assert_read(Type::Status, ResponseType::Ok) .await @@ -77,6 +97,9 @@ pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { // Delete folder and expect an update imap.send("DELETE Provolone").await; imap.assert_read(Type::Tagged, ResponseType::Ok).await; + if is_cluster_test { + tokio::time::sleep(SLEEP).await; + } imap_check .assert_read(Type::Status, ResponseType::Ok) .await @@ -88,6 +111,9 @@ pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { imap.assert_read(Type::Continuation, ResponseType::Ok).await; imap.send_untagged(message).await; imap.assert_read(Type::Tagged, ResponseType::Ok).await; + if is_cluster_test { + tokio::time::sleep(SLEEP).await; + } imap_check .assert_read(Type::Status, ResponseType::Ok) .await @@ -108,6 +134,9 @@ pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { imap.send("STORE 1 +FLAGS (\\Deleted)").await; imap.assert_read(Type::Tagged, ResponseType::Ok).await; + if is_cluster_test { + tokio::time::sleep(SLEEP).await; + } imap_check .assert_read(Type::Status, ResponseType::Ok) .await @@ -118,6 +147,9 @@ pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { .await .assert_contains("* 1 EXPUNGE") .assert_contains("* 0 EXISTS"); + if is_cluster_test { + tokio::time::sleep(SLEEP).await; + } imap_check .assert_read(Type::Status, ResponseType::Ok) .await @@ -133,7 +165,7 @@ pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { .assert_contains("* 0 EXISTS"); // Test SMTP delivery notifications - let mut lmtp = SmtpConnection::connect_port(11201).await; + let mut lmtp = SmtpConnection::connect_port(if is_cluster_test { 17000 } else { 11201 }).await; lmtp.ingest( "bill@example.com", &["jdoe@example.com"], @@ -148,11 +180,18 @@ pub async fn test(imap: &mut ImapConnection, imap_check: &mut ImapConnection) { ), ) .await; + if is_cluster_test { + tokio::time::sleep(SLEEP).await; + } imap_check .assert_read(Type::Status, ResponseType::Ok) .await .assert_contains("STATUS \"INBOX\"") - .assert_contains("MESSAGES 11"); + .assert_contains(if is_cluster_test { + "MESSAGES 1" + } else { + "MESSAGES 11" + }); // Stop IDLE mode imap_check.send_raw("DONE").await; diff --git a/tests/src/imap/mod.rs b/tests/src/imap/mod.rs index 5e61a46d..db29d51d 100644 --- a/tests/src/imap/mod.rs +++ b/tests/src/imap/mod.rs @@ -26,6 +26,7 @@ use crate::{ use ::managesieve::core::ManageSieveSessionManager; use ::store::Stores; use ahash::AHashSet; +use base64::{Engine, engine::general_purpose}; use common::{ Caches, Core, Data, Inner, Server, config::{ @@ -95,7 +96,7 @@ pub async fn imap_tests() { store::test(&mut imap, &mut imap_check, &handle).await; copy_move::test(&mut imap, &mut imap_check).await; thread::test(&mut imap, &mut imap_check).await; - idle::test(&mut imap, &mut imap_check).await; + idle::test(&mut imap, &mut imap_check, false).await; condstore::test(&mut imap, &mut imap_check).await; acl::test(&mut imap, &mut imap_check).await; @@ -169,7 +170,7 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest { let cache = Caches::parse(&mut config); let store = core.storage.data.clone(); - let (ipc, mut ipc_rxs) = build_ipc(&mut config); + let (ipc, mut ipc_rxs) = build_ipc(&mut config, false); let inner = Arc::new(Inner { shared_core: core.into_shared(), data, @@ -306,8 +307,11 @@ pub enum Type { impl ImapConnection { pub async fn connect(tag: &'static [u8]) -> Self { - let (reader, writer) = - tokio::io::split(TcpStream::connect("127.0.0.1:9991").await.unwrap()); + Self::connect_to(tag, "127.0.0.1:9991").await + } + + pub async fn connect_to(tag: &'static [u8], addr: impl AsRef) -> Self { + let (reader, writer) = tokio::io::split(TcpStream::connect(addr.as_ref()).await.unwrap()); ImapConnection { tag, reader: BufReader::new(reader).lines(), @@ -377,6 +381,16 @@ impl ImapConnection { } } + pub async fn authenticate(&mut self, user: &str, pass: &str) { + let creds = general_purpose::STANDARD.encode(format!("\0{user}\0{pass}")); + self.send(&format!( + "AUTHENTICATE PLAIN {{{}+}}\r\n{creds}", + creds.len() + )) + .await; + self.assert_read(Type::Tagged, ResponseType::Ok).await; + } + pub async fn send(&mut self, text: &str) { //let c = println!("-> {}{:?}", std::str::from_utf8(self.tag).unwrap(), text); self.writer.write_all(self.tag).await.unwrap(); diff --git a/tests/src/jmap/enterprise.rs b/tests/src/jmap/enterprise.rs index b1d5e118..9ee2b1c0 100644 --- a/tests/src/jmap/enterprise.rs +++ b/tests/src/jmap/enterprise.rs @@ -186,7 +186,7 @@ impl EnterpriseCore for Core { async fn alerts(server: &Server) { // Make sure the required metrics are set to 0 assert_eq!( - Collector::read_event_metric(EventType::Cluster(ClusterEvent::Error).id()), + Collector::read_event_metric(EventType::Cluster(ClusterEvent::PublisherError).id()), 0 ); assert_eq!(Collector::read_metric(MetricType::DomainCount), 0.0); @@ -196,12 +196,12 @@ async fn alerts(server: &Server) { ); // Increment metrics to trigger alerts - Collector::update_event_counter(EventType::Cluster(ClusterEvent::Error), 5); + Collector::update_event_counter(EventType::Cluster(ClusterEvent::PublisherError), 5); Collector::update_gauge(MetricType::DomainCount, 3); // Make sure the values were set assert_eq!( - Collector::read_event_metric(EventType::Cluster(ClusterEvent::Error).id()), + Collector::read_event_metric(EventType::Cluster(ClusterEvent::PublisherError).id()), 5 ); assert_eq!(Collector::read_metric(MetricType::DomainCount), 3.0); diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 490f3e40..20d86376 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -73,7 +73,6 @@ pub mod purge; pub mod push_subscription; pub mod quota; pub mod sieve_script; -pub mod stress_test; pub mod thread_get; pub mod thread_merge; pub mod vacation_response; @@ -135,19 +134,6 @@ async fn jmap_tests_() { } } -#[tokio::test(flavor = "multi_thread")] -#[ignore] -pub async fn jmap_stress_tests() { - let params = init_jmap_tests( - &std::env::var("STORE") - .expect("Missing store type. Try running `STORE= cargo test`"), - true, - ) - .await; - stress_test::test(params.server.clone(), params.client).await; - params.temp_dir.delete(); -} - #[ignore] #[tokio::test(flavor = "multi_thread")] pub async fn jmap_metric_tests() { @@ -328,7 +314,7 @@ async fn init_jmap_tests(store_id: &str, delete_if_exists: bool) -> JMAPTest { let data = Data::parse(&mut config); let cache = Caches::parse(&mut config); let store = core.storage.data.clone(); - let (ipc, mut ipc_rxs) = build_ipc(&mut config); + let (ipc, mut ipc_rxs) = build_ipc(&mut config, false); let inner = Arc::new(Inner { shared_core: core.into_shared(), data, diff --git a/tests/src/lib.rs b/tests/src/lib.rs index c8a05b99..005c728d 100644 --- a/tests/src/lib.rs +++ b/tests/src/lib.rs @@ -15,6 +15,8 @@ use trc::Collector; #[global_allocator] static GLOBAL: Jemalloc = Jemalloc; +#[cfg(test)] +pub mod cluster; #[cfg(test)] pub mod directory; #[cfg(test)] @@ -76,3 +78,16 @@ pub fn enable_logging() { } } } + +pub const TEST_USERS: &[(&str, &str, &str, &str)] = &[ + ("admin", "secret1", "Superuser", "admin@example,com"), + ("john", "secret2", "John Doe", "jdoe@example.com"), + ( + "jane", + "secret3", + "Jane Doe-Smith", + "jane.smith@example.com", + ), + ("bill", "secret4", "Bill Foobar", "bill@example,com"), + ("mike", "secret5", "Mike Noquota", "mike@example,com"), +]; diff --git a/tests/src/smtp/mod.rs b/tests/src/smtp/mod.rs index 71433677..f1f44d7e 100644 --- a/tests/src/smtp/mod.rs +++ b/tests/src/smtp/mod.rs @@ -173,7 +173,7 @@ impl TestSMTP { } pub fn inner_with_rxs(&self) -> (Arc, IpcReceivers) { - let (ipc, ipc_rxs) = build_ipc(&mut Config::default()); + let (ipc, ipc_rxs) = build_ipc(&mut Config::default(), false); ( Inner { @@ -191,7 +191,7 @@ impl TestSMTP { let store = core.storage.data.clone(); let blob_store = core.storage.blob.clone(); let shared_core = core.into_shared(); - let (ipc, mut ipc_rxs) = build_ipc(&mut Config::default()); + let (ipc, mut ipc_rxs) = build_ipc(&mut Config::default(), false); TestSMTP { queue_receiver: QueueReceiver { diff --git a/tests/src/webdav/mod.rs b/tests/src/webdav/mod.rs index 2c7688b5..0ed8b188 100644 --- a/tests/src/webdav/mod.rs +++ b/tests/src/webdav/mod.rs @@ -5,7 +5,7 @@ */ use crate::{ - AssertConfig, add_test_certs, directory::internal::TestInternalDirectory, + AssertConfig, TEST_USERS, add_test_certs, directory::internal::TestInternalDirectory, jmap::assert_is_empty, store::TempDir, }; use ::managesieve::core::ManageSieveSessionManager; @@ -133,7 +133,7 @@ async fn init_webdav_tests(store_id: &str, delete_if_exists: bool) -> WebDavTest let cache = Caches::parse(&mut config); let store = core.storage.data.clone(); - let (ipc, mut ipc_rxs) = build_ipc(&mut config); + let (ipc, mut ipc_rxs) = build_ipc(&mut config, false); let inner = Arc::new(Inner { shared_core: core.into_shared(), data, @@ -194,7 +194,7 @@ async fn init_webdav_tests(store_id: &str, delete_if_exists: bool) -> WebDavTest // Create test accounts let mut clients = AHashMap::new(); - for (account, secret, name, email) in TEST_DAV_USERS { + for (account, secret, name, email) in TEST_USERS { let account_id = store .create_test_user(account, secret, name, &[email]) .await; @@ -1127,16 +1127,3 @@ ansi = true disabled-events = ["network.*"] "#; - -pub const TEST_DAV_USERS: &[(&str, &str, &str, &str)] = &[ - ("admin", "secret1", "Superuser", "admin@example,com"), - ("john", "secret2", "John Doe", "jdoe@example.com"), - ( - "jane", - "secret3", - "Jane Doe-Smith", - "jane.smith@example.com", - ), - ("bill", "secret4", "Bill Foobar", "bill@example,com"), - ("mike", "secret5", "Mike Noquota", "mike@example,com"), -]; diff --git a/tests/src/webdav/principals.rs b/tests/src/webdav/principals.rs index 5cf67686..740978b0 100644 --- a/tests/src/webdav/principals.rs +++ b/tests/src/webdav/principals.rs @@ -5,7 +5,7 @@ */ use super::WebDavTest; -use crate::webdav::{TEST_DAV_USERS, prop::ALL_DAV_PROPERTIES}; +use crate::{TEST_USERS, webdav::prop::ALL_DAV_PROPERTIES}; use dav_proto::schema::property::{DavProperty, PrincipalProperty, WebDavProperty}; use groupware::DavResourceName; use hyper::StatusCode; @@ -23,7 +23,7 @@ pub async fn test(test: &WebDavTest) { ALL_DAV_PROPERTIES, ) .await; - for (account, _, name, _) in TEST_DAV_USERS { + for (account, _, name, _) in TEST_USERS { let props = response.properties(&format!( "{}/{}/", DavResourceName::Principal.base_path(), @@ -174,7 +174,7 @@ pub async fn test(test: &WebDavTest) { .with_values([format!("D:href:{}/jane/", DavResourceName::Card.base_path()).as_str()]) .with_status(StatusCode::OK); - for (account, _, name, _) in TEST_DAV_USERS + for (account, _, name, _) in TEST_USERS .iter() .filter(|(account, _, _, _)| ["jane", "support"].contains(account)) { @@ -310,7 +310,7 @@ pub async fn test(test: &WebDavTest) { response .properties(&format!("{}/jane/", DavResourceName::Principal.base_path())) .get(DavProperty::WebDav(WebDavProperty::DisplayName)) - .with_values([TEST_DAV_USERS + .with_values([TEST_USERS .iter() .find(|(account, _, _, _)| *account == "jane") .unwrap()