diff --git a/Cargo.lock b/Cargo.lock index 2c8c3236..17729b7e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5731,9 +5731,9 @@ dependencies = [ [[package]] name = "socket2" -version = "0.5.6" +version = "0.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05ffd9c0a93b7543e062e759284fcf5f5e3b098501104bfbdde4d404db792871" +checksum = "ce305eb0b4296696835b71df73eb912e0f1ffd2556a501fcede6e0c50349191c" dependencies = [ "libc", "windows-sys 0.52.0", diff --git a/crates/common/src/addresses.rs b/crates/common/src/addresses.rs index 3e17d5b3..75fa1d01 100644 --- a/crates/common/src/addresses.rs +++ b/crates/common/src/addresses.rs @@ -1,3 +1,26 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + use std::borrow::Cow; use directory::Directory; diff --git a/crates/common/src/expr/eval.rs b/crates/common/src/expr/eval.rs index ccc1bcca..19bb3bff 100644 --- a/crates/common/src/expr/eval.rs +++ b/crates/common/src/expr/eval.rs @@ -576,24 +576,6 @@ impl Display for Variable<'_> { } } -trait IntoBool { - fn into_bool(self) -> bool; -} - -impl IntoBool for f64 { - #[inline(always)] - fn into_bool(self) -> bool { - self != 0.0 - } -} - -impl IntoBool for i64 { - #[inline(always)] - fn into_bool(self) -> bool { - self != 0 - } -} - impl<'x> From<&'x Constant> for Variable<'x> { fn from(value: &'x Constant) -> Self { match value { diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 9973b396..24adadba 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -1,3 +1,26 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + use std::{borrow::Cow, net::IpAddr, sync::Arc}; use arc_swap::ArcSwap; diff --git a/crates/common/src/listener/blocked.rs b/crates/common/src/listener/blocked.rs index 9da78d79..761b3e54 100644 --- a/crates/common/src/listener/blocked.rs +++ b/crates/common/src/listener/blocked.rs @@ -21,7 +21,7 @@ * for more details. */ -use std::{fmt::Debug, net::IpAddr}; +use std::{fmt::Debug, net::IpAddr, sync::atomic::AtomicU8}; use ahash::AHashSet; use parking_lot::RwLock; @@ -35,6 +35,7 @@ use crate::Core; pub struct BlockedIps { pub ip_addresses: RwLock>, + pub version: AtomicU8, ip_networks: Vec, has_networks: bool, limiter_rate: Option, @@ -71,6 +72,7 @@ impl BlockedIps { has_networks: !ip_networks.is_empty(), ip_networks, limiter_rate: config.property_or_default::("authentication.fail2ban", "100/1d"), + version: 0.into(), } } } @@ -103,6 +105,9 @@ impl Core { }]) .await?; + // Increment version + self.network.blocked_ips.increment_version(); + return Ok(true); } } @@ -126,6 +131,13 @@ impl Core { } } +impl BlockedIps { + pub fn increment_version(&self) { + self.version + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } +} + impl Default for BlockedIps { fn default() -> Self { Self { @@ -133,6 +145,7 @@ impl Default for BlockedIps { ip_networks: Default::default(), has_networks: Default::default(), limiter_rate: Default::default(), + version: Default::default(), } } } @@ -144,6 +157,10 @@ impl Clone for BlockedIps { ip_networks: self.ip_networks.clone(), has_networks: self.has_networks, limiter_rate: self.limiter_rate.clone(), + version: self + .version + .load(std::sync::atomic::Ordering::Relaxed) + .into(), } } } diff --git a/crates/common/src/listener/listen.rs b/crates/common/src/listener/listen.rs index df4fac2a..f9a8273b 100644 --- a/crates/common/src/listener/listen.rs +++ b/crates/common/src/listener/listen.rs @@ -325,7 +325,7 @@ impl Servers { pub fn spawn( mut self, spawn: impl Fn(Server, TcpAcceptor, watch::Receiver), - ) -> watch::Sender { + ) -> (watch::Sender, watch::Receiver) { // Spawn listeners let (shutdown_tx, shutdown_rx) = watch::channel(false); for server in self.servers { @@ -336,7 +336,7 @@ impl Servers { spawn(server, acceptor, shutdown_rx.clone()); } - shutdown_tx + (shutdown_tx, shutdown_rx) } } diff --git a/crates/common/src/manager/boot.rs b/crates/common/src/manager/boot.rs index fb115283..1692c14f 100644 --- a/crates/common/src/manager/boot.rs +++ b/crates/common/src/manager/boot.rs @@ -221,6 +221,22 @@ impl BootManager { ))); } + // Generate a Cluster encryption key if missing + if config + .value("cluster.key") + .filter(|v| !v.is_empty()) + .is_none() + { + insert_keys.push(ConfigKey::from(( + "cluster.key", + thread_rng() + .sample_iter(Alphanumeric) + .take(64) + .map(char::from) + .collect::(), + ))); + } + // Download SPAM filters if missing if config .value("version.spam-filter") diff --git a/crates/jmap/src/api/http.rs b/crates/jmap/src/api/http.rs index fb10f455..ff39072a 100644 --- a/crates/jmap/src/api/http.rs +++ b/crates/jmap/src/api/http.rs @@ -49,7 +49,7 @@ use crate::{ auth::oauth::OAuthMetadata, blob::{DownloadResponse, UploadResponse}, services::state, - JMAP, + JmapInstance, JMAP, }; use super::{HtmlResponse, HttpRequest, HttpResponse, JmapSessionManager, JsonResponse}; @@ -356,7 +356,9 @@ impl JMAP { } RequestError::not_found().into_http_response() } +} +impl JmapInstance { async fn handle_session(self, session: SessionData) { let span = session.span; let _in_flight = session.in_flight; @@ -367,7 +369,7 @@ impl JMAP { .serve_connection( TokioIo::new(session.stream), service_fn(|req: hyper::Request| { - let jmap = self.clone(); + let jmap_instance = self.clone(); let span = span.clone(); let instance = session.instance.clone(); @@ -377,6 +379,7 @@ impl JMAP { event = "request", uri = req.uri().to_string(), ); + let jmap = JMAP::from(jmap_instance); // Obtain remote IP let remote_ip = if !jmap.core.jmap.http_use_forwarded { @@ -442,7 +445,7 @@ impl SessionManager for JmapSessionManager { self, session: SessionData, ) -> impl std::future::Future + Send { - JMAP::from(self.inner).handle_session(session) + self.inner.handle_session(session) } #[allow(clippy::manual_async_fn)] diff --git a/crates/jmap/src/api/management/reload.rs b/crates/jmap/src/api/management/reload.rs index 2cdd323d..053fc52f 100644 --- a/crates/jmap/src/api/management/reload.rs +++ b/crates/jmap/src/api/management/reload.rs @@ -28,6 +28,7 @@ use utils::url_params::UrlParams; use crate::{ api::{http::ToHttpResponse, HttpRequest, HttpResponse, JsonResponse}, + services::housekeeper::Event, JMAP, }; @@ -59,10 +60,15 @@ impl JMAP { }, (Some("server.blocked-ip"), &Method::GET) => { match self.core.reload_blocked_ips().await { - Ok(result) => JsonResponse::new(json!({ - "data": result.config, - })) - .into_http_response(), + Ok(result) => { + // Increment version counter + self.core.network.blocked_ips.increment_version(); + + JsonResponse::new(json!({ + "data": result.config, + })) + .into_http_response() + } Err(err) => err.into_http_response(), } } @@ -70,9 +76,22 @@ impl JMAP { match self.core.reload().await { Ok(result) => { if !UrlParams::new(req.uri().query()).has_key("dry-run") { - // Update core if let Some(core) = result.new_core { + // Update core self.shared_core.store(core.into()); + + // Increment version counter + self.inner.increment_config_version(); + } + + // Reload ACME + if let Err(err) = + self.inner.housekeeper_tx.send(Event::AcmeReload).await + { + tracing::warn!( + "Failed to send ACME reload event to housekeeper: {}", + err + ); } } diff --git a/crates/jmap/src/lib.rs b/crates/jmap/src/lib.rs index 93602f77..004b7801 100644 --- a/crates/jmap/src/lib.rs +++ b/crates/jmap/src/lib.rs @@ -21,7 +21,12 @@ * for more details. */ -use std::{collections::hash_map::RandomState, fmt::Display, sync::Arc, time::Duration}; +use std::{ + collections::hash_map::RandomState, + fmt::Display, + sync::{atomic::AtomicU8, Arc}, + time::Duration, +}; use auth::{rate_limit::ConcurrencyLimiters, AccessToken}; use common::{manager::webadmin::WebAdminManager, Core, DeliveryEvent, SharedCore}; @@ -100,6 +105,7 @@ pub struct Inner { pub access_tokens: TtlDashMap>, pub snowflake_id: SnowflakeIdGenerator, pub webadmin: WebAdminManager, + pub config_version: AtomicU8, pub concurrency_limiter: DashMap>, @@ -150,6 +156,7 @@ impl JMAP { cache_threads: LruCache::with_capacity( config.property("cache.thread.size").unwrap_or(2048), ), + config_version: 0.into(), }; // Unpack webadmin @@ -569,6 +576,13 @@ impl JMAP { } } +impl Inner { + pub fn increment_config_version(&self) { + self.config_version + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } +} + impl From for JMAP { fn from(value: JmapInstance) -> Self { let shared_core = value.core.clone(); diff --git a/crates/jmap/src/services/gossip/heartbeat.rs b/crates/jmap/src/services/gossip/heartbeat.rs new file mode 100644 index 00000000..ec7aeba2 --- /dev/null +++ b/crates/jmap/src/services/gossip/heartbeat.rs @@ -0,0 +1,132 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use super::{Peer, State, HEARTBEAT_WINDOW, HEARTBEAT_WINDOW_MASK}; +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 => { + tracing::debug!("Peer {} is now alive.", self.addr); + self.state = State::Alive; + + // Do not count stale heartbeats. + return true; + } + State::Suspected => { + tracing::debug!("Suspected peer {} was confirmed alive.", self.addr); + self.state = State::Alive; + } + State::Left if is_direct_ping => { + tracing::debug!( + "Peer {} is back online after leaving the cluster.", + 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() + }; + + /*tracing::debug!( + "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 { + tracing::debug!("Peer {} is offline.", self.addr); + self.state = State::Offline; + false + } else if phi > HB_PHI_SUSPECT_THRESHOLD { + tracing::debug!("Peer {} is suspected to be offline.", self.addr); + self.state = State::Suspected; + true + } else { + true + } + } +} diff --git a/crates/jmap/src/services/gossip/leave.rs b/crates/jmap/src/services/gossip/leave.rs new file mode 100644 index 00000000..10eb3308 --- /dev/null +++ b/crates/jmap/src/services/gossip/leave.rs @@ -0,0 +1,58 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use crate::services::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 { + tracing::debug!("Peer {} is leaving the cluster.", local_peer.addr); + + local_peer.state = State::Left; + local_peer.epoch = peer.epoch; + + // Reload + self.request_reload(); + + break; + } + } + } + } +} diff --git a/crates/jmap/src/services/gossip/mod.rs b/crates/jmap/src/services/gossip/mod.rs new file mode 100644 index 00000000..4009fcbe --- /dev/null +++ b/crates/jmap/src/services/gossip/mod.rs @@ -0,0 +1,147 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +pub mod heartbeat; +pub mod leave; +pub mod peer; +pub mod ping; +pub mod request; +pub mod spawn; + +use serde::{Deserialize, Serialize}; +use std::{ + net::{IpAddr, SocketAddr}, + sync::atomic::Ordering, + time::Instant, +}; +use tokio::sync::mpsc; + +use crate::JmapInstance; + +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 core: JmapInstance, + 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, Serialize, Deserialize, 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 + .core + .jmap_inner + .config_version + .load(Ordering::Relaxed), + gen_lists: cluster + .core + .core + .load() + .network + .blocked_ips + .version + .load(Ordering::Relaxed), + } + } +} + +impl Gossiper { + pub async fn send_gossip(&self, dest: IpAddr, request: Request) { + if let Err(err) = self + .gossip_tx + .send((SocketAddr::new(dest, self.port), request)) + .await + { + tracing::error!("Failed to send gossip message: {}", err); + }; + } +} diff --git a/crates/jmap/src/services/gossip/peer.rs b/crates/jmap/src/services/gossip/peer.rs new file mode 100644 index 00000000..1e17ff0c --- /dev/null +++ b/crates/jmap/src/services/gossip/peer.rs @@ -0,0 +1,115 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::{fmt::Display, net::IpAddr, time::Instant}; + +use super::{Gossiper, Peer, PeerStatus, State, HEARTBEAT_WINDOW}; + +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/jmap/src/services/gossip/ping.rs b/crates/jmap/src/services/gossip/ping.rs new file mode 100644 index 00000000..b34c80a7 --- /dev/null +++ b/crates/jmap/src/services/gossip/ping.rs @@ -0,0 +1,211 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use smtp::queue; + +use crate::services::housekeeper; + +use super::{request::Request, Gossiper, PeerStatus}; + +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 core = self.core.clone(); + + tokio::spawn(async move { + tracing::debug!("One or more nodes became offline, reloading queues."); + + let _ = core + .jmap_inner + .housekeeper_tx + .send(housekeeper::Event::IndexStart) + .await; + let _ = core.smtp_inner.queue_tx.send(queue::Event::Reload).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() { + tracing::debug!("Received empty ping packet."); + 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 { + tracing::debug!( + "Peer {} has configuration changes.", + peer.addr + ); + update_config = true; + } + } + if local_peer.gen_lists != peer.gen_lists { + local_peer.gen_lists = peer.gen_lists; + if local_peer.hb_sum > 0 { + tracing::debug!("Peer {} has list changes.", peer.addr); + update_lists = true; + } + } + } + + continue 'outer; + } + } else if !remove_seeds { + remove_seeds = true; + } + } + + // Add new peer to the list. + tracing::info!("Discovered new peer at {}.", 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 core = self.core.core.clone(); + let inner = self.core.jmap_inner.clone(); + + tokio::spawn(async move { + let result = if update_config { + core.load().reload().await + } else { + core.load().reload_blocked_ips().await + }; + match result { + Ok(result) => { + if let Some(new_core) = result.new_core { + // Update core + core.store(new_core.into()); + + // Reload ACME + if let Err(err) = inner + .housekeeper_tx + .send(housekeeper::Event::AcmeReload) + .await + { + tracing::warn!( + "Failed to send ACME reload event to housekeeper: {}", + err + ); + } + } + } + Err(err) => { + tracing::error!("Failed to reload configuration: {}", err); + } + } + }); + } + } +} diff --git a/crates/jmap/src/services/gossip/request.rs b/crates/jmap/src/services/gossip/request.rs new file mode 100644 index 00000000..bfdc9b7a --- /dev/null +++ b/crates/jmap/src/services/gossip/request.rs @@ -0,0 +1,132 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use crate::auth::SymmetricEncrypt; + +use super::{EpochId, PeerStatus}; + +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/jmap/src/services/gossip/spawn.rs b/crates/jmap/src/services/gossip/spawn.rs new file mode 100644 index 00000000..8c60b0fb --- /dev/null +++ b/crates/jmap/src/services/gossip/spawn.rs @@ -0,0 +1,206 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use crate::auth::SymmetricEncrypt; +use crate::services::IPC_CHANNEL_BUFFER; +use crate::JmapInstance; + +use super::request::Request; +use super::{Gossiper, Peer, UDP_MAX_PAYLOAD}; +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.ping-interval", "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, core: JmapInstance, 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) => { + tracing::error!("Failed to bind UDP socket on '{}': {}", self.bind_addr, e); + return; + } + }, + nonce: b"428934328968".to_vec(), + encryptor: SymmetricEncrypt::new( + self.cluster_key.as_bytes(), + "gossipmonger context key", + ), + }); + tracing::info!( + bind.ip = self.bind_addr.to_string().as_str(), + bind.port = self.port, + "Starting gossip service" + ); + + // 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, + core, + gossip_tx, + }; + let quidnunc_ = quidnunc.clone(); + + // 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 { + tracing::error!( + "Failed to send UDP packet to {}: {}", + target_addr, + err + ); + } + } + Err(err) => { + tracing::error!("Failed to encrypt UDP packet to {}: {}", target_addr, err); + } + } + } + }); + + // 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) { + //tracing::debug!("Received packet from {}", addr); + 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 { + tracing::debug!("Received invalid gossip message from {}", addr); + } + }, + Err(err) => { + tracing::debug!("Failed to decrypt UDP packet from {}: {}", addr, err); + }, + } + } + Err(e) => { + tracing::error!("Gossip process ended, socket.recv_from() failed: {}", e); + } + } + }, + _ = tokio::time::sleep(wait) => { + // Send ping + gossiper.ping_peers().await; + last_ping = Instant::now(); + }, + _ = shutdown_rx.changed() => { + tracing::debug!("Gossip listener shutting down."); + + // Broadcast leave message + gossiper.broadcast_leave().await; + + break; + } + }; + + // Calculate next ping interval + wait = ping_interval.saturating_sub(last_ping.elapsed()); + } + }); + } +} diff --git a/crates/jmap/src/services/housekeeper.rs b/crates/jmap/src/services/housekeeper.rs index ad4fd383..189a7d49 100644 --- a/crates/jmap/src/services/housekeeper.rs +++ b/crates/jmap/src/services/housekeeper.rs @@ -37,6 +37,7 @@ use super::IPC_CHANNEL_BUFFER; pub enum Event { IndexStart, IndexDone, + AcmeReload, AcmeReschedule { provider_id: String, renew_at: Instant, @@ -113,11 +114,41 @@ pub fn spawn_housekeeper(core: JmapInstance, mut rx: mpsc::Receiver) { loop { match tokio::time::timeout(queue.wake_up_time(), rx.recv()).await { Ok(Some(event)) => match event { + Event::AcmeReload => { + let core_ = core.core.load().clone(); + let inner = core.jmap_inner.clone(); + + tokio::spawn(async move { + for provider in core_.tls.acme_providers.values() { + match core_.init_acme(provider).await { + Ok(renew_at) => { + inner + .housekeeper_tx + .send(Event::AcmeReschedule { + provider_id: provider.id.clone(), + renew_at: Instant::now() + renew_at, + }) + .await + .ok(); + } + Err(err) => { + tracing::error!( + context = "acme", + event = "error", + error = ?err, + "Failed to reload ACME certificate manager."); + } + }; + } + }); + } Event::AcmeReschedule { provider_id, renew_at, } => { - queue.schedule(renew_at, ActionClass::Acme(provider_id)); + let action = ActionClass::Acme(provider_id); + queue.remove_action(&action); + queue.schedule(renew_at, action); } Event::IndexStart => { if !index_busy { @@ -192,6 +223,8 @@ pub fn spawn_housekeeper(core: JmapInstance, mut rx: mpsc::Receiver) { } }; + inner.increment_config_version(); + inner .housekeeper_tx .send(Event::AcmeReschedule { @@ -267,6 +300,10 @@ impl Queue { self.heap.push(Action { due, event }); } + pub fn remove_action(&mut self, event: &ActionClass) { + self.heap.retain(|e| &e.event != event); + } + pub fn wake_up_time(&self) -> Duration { self.heap .peek() diff --git a/crates/jmap/src/services/mod.rs b/crates/jmap/src/services/mod.rs index 405cfc47..21051a29 100644 --- a/crates/jmap/src/services/mod.rs +++ b/crates/jmap/src/services/mod.rs @@ -22,6 +22,7 @@ */ pub mod delivery; +pub mod gossip; pub mod housekeeper; pub mod index; pub mod ingest; diff --git a/crates/main/src/main.rs b/crates/main/src/main.rs index a52bfd0f..9b796a29 100644 --- a/crates/main/src/main.rs +++ b/crates/main/src/main.rs @@ -25,7 +25,11 @@ use std::time::Duration; use common::{config::server::ServerProtocol, manager::boot::BootManager}; use imap::core::{ImapSessionManager, IMAP}; -use jmap::{api::JmapSessionManager, services::IPC_CHANNEL_BUFFER, JMAP}; +use jmap::{ + api::JmapSessionManager, + services::{gossip::spawn::GossiperBuilder, IPC_CHANNEL_BUFFER}, + JMAP, +}; use managesieve::core::ManageSieveSessionManager; use smtp::core::{SmtpSessionManager, SMTP}; use tokio::sync::mpsc; @@ -52,13 +56,14 @@ async fn main() -> std::io::Result<()> { let smtp = SMTP::init(&mut config, core.clone(), delivery_tx).await; let jmap = JMAP::init(&mut config, delivery_rx, core.clone(), smtp.inner.clone()).await; let imap = IMAP::init(&mut config, jmap.clone()).await; + let gossiper = GossiperBuilder::try_parse(&mut config); // Log configuration errors config.log_errors(init.guards.is_none()); config.log_warnings(init.guards.is_none()); // Spawn servers - let shutdown_tx = init.servers.spawn(|server, acceptor, shutdown_rx| { + let (shutdown_tx, shutdown_rx) = init.servers.spawn(|server, acceptor, shutdown_rx| { match &server.protocol { ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn( SmtpSessionManager::new(smtp.clone()), @@ -87,6 +92,11 @@ async fn main() -> std::io::Result<()> { }; }); + // Spawn gossip + if let Some(gossiper) = gossiper { + gossiper.spawn(jmap, shutdown_rx).await; + } + // Wait for shutdown signal wait_for_shutdown(&format!( "Shutting down Stalwart Mail Server v{}...", diff --git a/tests/resources/scripts/create_test_cluster.sh b/tests/resources/scripts/create_test_cluster.sh new file mode 100644 index 00000000..4729d1ac --- /dev/null +++ b/tests/resources/scripts/create_test_cluster.sh @@ -0,0 +1,53 @@ +#!/bin/bash + +BASE_DIR="/Users/me/Downloads/stalwart-cluster" +FEATURES="rocks" +NUM_NODES=5 + +# Kill previous processes +sudo pkill stalwart-mail + +# Delete previous tests +rm -rf $BASE_DIR + +# Build the mail-server binary +cargo build -p mail-server --no-default-features --features "$FEATURES" + +for NUM in $(seq 1 $NUM_NODES); do + sudo ifconfig en0 alias 10.0.$NUM.1 netmask 255.255.255.0 + mkdir -p $BASE_DIR/data$NUM + cat < $BASE_DIR/config$NUM.toml +cluster.bind-addr = "10.0._N_.1" +cluster.key = "the cluster key" +cluster.seed-nodes = ["10.0.1.1", "10.0.2.1", "10.0.3.1"] +authentication.fallback-admin.secret = "secret" +authentication.fallback-admin.user = "admin" +directory.internal.store = "rocksdb" +directory.internal.type = "internal" +lookup.default.hostname = "mail_N_.example.org" +server.http.permissive-cors = true +server.listener.https.bind = "10.0._N_.1:1443" +server.listener.https.protocol = "http" +server.listener.https.tls.implicit = true +server.listener.imap.bind = "10.0._N_.1:1143" +server.listener.imap.protocol = "imap" +server.listener.smtp.bind = "10.0._N_.1:1125" +server.listener.smtp.protocol = "smtp" +storage.blob = "rocksdb" +storage.data = "rocksdb" +storage.directory = "internal" +storage.fts = "rocksdb" +storage.lookup = "rocksdb" +store.rocksdb.compression = "lz4" +store.rocksdb.path = "_D_/data_N_" +store.rocksdb.type = "rocksdb" +tracer.stdout.ansi = true +tracer.stdout.enable = true +tracer.stdout.level = "debug" +tracer.stdout.type = "stdout" +config.resource.spam-filter = "file:///dev/null" +config.resource.webadmin = "file:///dev/null" +EOF + + sudo ./target/debug/stalwart-mail --config $BASE_DIR/config$NUM.toml & +done diff --git a/tests/src/imap/mod.rs b/tests/src/imap/mod.rs index 6e00673e..61894fc1 100644 --- a/tests/src/imap/mod.rs +++ b/tests/src/imap/mod.rs @@ -307,7 +307,7 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest { config.assert_no_errors(); // Spawn servers - let shutdown_tx = servers.spawn(|server, acceptor, shutdown_rx| { + let (shutdown_tx, _) = servers.spawn(|server, acceptor, shutdown_rx| { match &server.protocol { ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn( SmtpSessionManager::new(smtp.clone()), diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 4e1b5ff3..cbf8aaa3 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -451,7 +451,7 @@ async fn init_jmap_tests(store_id: &str, delete_if_exists: bool) -> JMAPTest { config.assert_no_errors(); // Spawn servers - let shutdown_tx = servers.spawn(|server, acceptor, shutdown_rx| { + let (shutdown_tx, _) = servers.spawn(|server, acceptor, shutdown_rx| { match &server.protocol { ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn( SmtpSessionManager::new(smtp.clone()), diff --git a/tests/src/smtp/outbound/mod.rs b/tests/src/smtp/outbound/mod.rs index 17d7c70b..577345f2 100644 --- a/tests/src/smtp/outbound/mod.rs +++ b/tests/src/smtp/outbound/mod.rs @@ -159,25 +159,27 @@ impl TestServer { let jmap_manager = JmapSessionManager::new(jmap); config.assert_no_errors(); - servers.spawn(|server, acceptor, shutdown_rx| { - match &server.protocol { - ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn( - smtp_manager.clone(), - instance.core.clone(), - acceptor, - shutdown_rx, - ), - ServerProtocol::Http => server.spawn( - jmap_manager.clone(), - instance.core.clone(), - acceptor, - shutdown_rx, - ), - ServerProtocol::Imap | ServerProtocol::ManageSieve => { - unreachable!() - } - }; - }) + servers + .spawn(|server, acceptor, shutdown_rx| { + match &server.protocol { + ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn( + smtp_manager.clone(), + instance.core.clone(), + acceptor, + shutdown_rx, + ), + ServerProtocol::Http => server.spawn( + jmap_manager.clone(), + instance.core.clone(), + acceptor, + shutdown_rx, + ), + ServerProtocol::Imap | ServerProtocol::ManageSieve => { + unreachable!() + } + }; + }) + .0 } pub fn new_session(&self) -> Session {