Cluster: Broadcast MTA queue refresh events to all nodes

crates/common/src/ipc.rs

@@ -78,6 +78,7 @@ pub enum BroadcastEvent {
     CacheInvalidateAll,
     CacheInvalidateNegative,
     MtaQueueStatus { is_running: bool },
+    QueueRefresh,
 }
 
 #[derive(Debug, Clone, Copy)]

crates/services/src/broadcast/mod.rs

@@ -135,6 +135,9 @@ impl BroadcastBatch<Vec<BroadcastEvent>> {
                         serialized.push(11u8);
                     }
                 }
+                BroadcastEvent::QueueRefresh => {
+                    serialized.push(12u8);
+                }
             }
         }
         serialized
@@ -266,6 +269,7 @@ where
                 9 => Ok(Some(BroadcastEvent::CacheInvalidateNegative)),
                 10 => Ok(Some(BroadcastEvent::MtaQueueStatus { is_running: true })),
                 11 => Ok(Some(BroadcastEvent::MtaQueueStatus { is_running: false })),
+                12 => Ok(Some(BroadcastEvent::QueueRefresh)),
                 _ => Err(()),
             }
         } else {

crates/services/src/broadcast/subscriber.rs

@@ -153,6 +153,15 @@ pub fn spawn_broadcast_subscriber(inner: Arc<Inner>, mut shutdown_rx: watch::Rec
                                                         .send(QueueEvent::Paused(!is_running))
                                                         .await;
                                             }
+                                            BroadcastEvent::QueueRefresh => {
+                                                if inner.shared_core.load().network.roles.outbound_mta {
+                                                    let _ = inner
+                                                            .ipc
+                                                            .queue_tx
+                                                            .send(QueueEvent::Refresh)
+                                                            .await;
+                                                }
+                                            }
                                             BroadcastEvent::RegistryChange(change) => {
                                                 match Box::pin(inner.build_server().reload_registry(change)).await {
                                                     Ok(result) => {
@@ -255,5 +264,6 @@ fn log_event(event: &BroadcastEvent) -> trc::Value {
                 "MtaQueuePaused".into()
             }
         }
+        BroadcastEvent::QueueRefresh => "QueueRefresh".into(),
     }
 }

crates/smtp/src/queue/spool.rs

@@ -17,7 +17,7 @@ use crate::queue::{
 use ahash::{AHashMap, AHashSet};
 use common::config::smtp::auth::DkimSigners;
 use common::config::smtp::queue::{ArchivedQueueExpiry, QueueName};
-use common::ipc::QueueEvent;
+use common::ipc::{BroadcastEvent, QueueEvent};
 use common::network::RcptResolution;
 use common::{KV_LOCK_QUEUE_MESSAGE, Server};
 use mail_auth::AuthenticatedMessage;
@@ -601,6 +601,8 @@ impl MessageWrapper {
             );
         }
 
+        server.cluster_broadcast(BroadcastEvent::QueueRefresh).await;
+
         true
     }