Gossip service implementation for cluster node autodiscovery with failure detection

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<T: SessionStream>(self, session: SessionData<T>) {
         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<body::Incoming>| {
-                    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<T>,
     ) -> impl std::future::Future<Output = ()> + Send {
-        JMAP::from(self.inner).handle_session(session)
+        self.inner.handle_session(session)
     }
 
     #[allow(clippy::manual_async_fn)]

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
+                                );
                             }
                         }
 

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<u32, Arc<AccessToken>>,
     pub snowflake_id: SnowflakeIdGenerator,
     pub webadmin: WebAdminManager,
+    pub config_version: AtomicU8,
 
     pub concurrency_limiter: DashMap<u32, Arc<ConcurrencyLimiters>>,
 
@@ -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<JmapInstance> for JMAP {
     fn from(value: JmapInstance) -> Self {
         let shared_core = value.core.clone();

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 <http://www.gnu.org/licenses/>.
+ *
+ * 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
+        }
+    }
+}

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 <http://www.gnu.org/licenses/>.
+ *
+ * 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<PeerStatus> = 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<PeerStatus>) {
+        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;
+                }
+            }
+        }
+    }
+}

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 <http://www.gnu.org/licenses/>.
+ *
+ * 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<Peer>,
+    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<u32>,
+    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);
+        };
+    }
+}

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 <http://www.gnu.org/licenses/>.
+ *
+ * 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<PeerStatus> {
+        let mut result: Vec<PeerStatus> = Vec::with_capacity(self.peers.len() + 1);
+        result.push(self.into());
+        for peer in self.peers.iter() {
+            result.push(peer.into());
+        }
+        result
+    }
+}
+
+impl From<PeerStatus> 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)
+    }
+}

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 <http://www.gnu.org/licenses/>.
+ *
+ * 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<PeerStatus>, 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);
+                    }
+                }
+            });
+        }
+    }
+}

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 <http://www.gnu.org/licenses/>.
+ *
+ * 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<PeerStatus>),
+    Pong(Vec<PeerStatus>),
+    Leave(Vec<PeerStatus>),
+}
+
+impl Request {
+    const PING: u8 = 0;
+    const PONG: u8 = 1;
+    const LEAVE: u8 = 2;
+
+    pub fn from_bytes(bytes: &[u8]) -> Option<Request> {
+        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::<PeerStatus>());
+        '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<u8> {
+        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::<usize>()
+                + (peers.len() * std::mem::size_of::<PeerStatus>())
+                + 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
+    }
+}

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 <http://www.gnu.org/licenses/>.
+ *
+ * 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<Peer>,
+    ping_interval: Duration,
+}
+
+//  Quidnunc: an inquisitive and gossipy person, from Latin quid nunc? 'what now?'.
+struct Quidnunc {
+    socket: UdpSocket,
+    nonce: Vec<u8>,
+    encryptor: SymmetricEncrypt,
+}
+
+impl GossiperBuilder {
+    pub fn try_parse(config: &mut Config) -> Option<Self> {
+        // Load configuration
+        let bind_addr = config.property::<IpAddr>("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::<IpAddr>("cluster.advertise-addr")
+                .unwrap_or(bind_addr),
+            port: config
+                .property_or_default::<u16>("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::<IpAddr>("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<bool>) {
+        // 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());
+            }
+        });
+    }
+}

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<Event>) {
         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<Event>) {
                                             }
                                         };
 
+                                        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()

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;