kumod: refresh queue_config periodically

This commit causes the scheduled queue maintainer to refresh
the queue config by calling the get_queue_config event approximately
every minute while the queue is alive.

In addition, we now thread the routing_domain through to get_queue_config

crates/kumod/src/egress_source.rs

@@ -4,6 +4,7 @@ use crate::ready_queue::{ReadyQueueManager, ReadyQueueName};
 use anyhow::Context;
 use config::LuaConfig;
 use gcd::Gcd;
+use kumo_server_common::config_handle::ConfigHandle;
 use lruttl::LruCacheWithTtl;
 use mlua::prelude::LuaUserData;
 use serde::{Deserialize, Serialize};
@@ -360,7 +361,11 @@ impl EgressPoolRoundRobin {
         }
     }
 
-    pub async fn next(&self, queue_name: &str, queue_config: &QueueConfig) -> RoundRobinResult {
+    pub async fn next(
+        &self,
+        queue_name: &str,
+        queue_config: &ConfigHandle<QueueConfig>,
+    ) -> RoundRobinResult {
         if self.entries.is_empty() {
             return RoundRobinResult::NoSources;
         }

crates/kumod/src/queue.rs

@@ -8,7 +8,8 @@ use crate::smtp_dispatcher::SmtpProtocol;
 use crate::spool::SpoolManager;
 use anyhow::Context;
 use chrono::Utc;
-use config::load_config;
+use config::{load_config, LuaConfig};
+use kumo_server_common::config_handle::ConfigHandle;
 use kumo_server_lifecycle::{Activity, ShutdownSubcription};
 use kumo_server_runtime::{rt_spawn, spawn, spawn_blocking};
 use message::message::QueueNameComponents;
@@ -374,24 +375,37 @@ pub struct Queue {
     name: String,
     queue: TimeQ<Message>,
     last_change: Instant,
-    queue_config: QueueConfig,
+    queue_config: ConfigHandle<QueueConfig>,
     delayed_gauge: IntGauge,
     activity: Activity,
     rr: EgressPoolRoundRobin,
 }
 
 impl Queue {
-    pub async fn new(name: String) -> anyhow::Result<QueueHandle> {
-        let mut config = load_config().await?;
-
+    async fn call_get_queue_config(
+        name: &str,
+        config: &mut LuaConfig,
+    ) -> anyhow::Result<QueueConfig> {
         let components = QueueNameComponents::parse(&name);
         let queue_config: QueueConfig = config
             .async_call_callback(
                 "get_queue_config",
-                (components.domain, components.tenant, components.campaign),
+                (
+                    components.domain,
+                    components.tenant,
+                    components.campaign,
+                    components.routing_domain,
+                ),
             )
             .await?;
 
+        Ok(queue_config)
+    }
+
+    pub async fn new(name: String) -> anyhow::Result<QueueHandle> {
+        let mut config = load_config().await?;
+        let queue_config = Self::call_get_queue_config(&name, &mut config).await?;
+
         let pool = EgressPool::resolve(queue_config.egress_pool.as_deref(), &mut config).await?;
         let rr = EgressPoolRoundRobin::new(&pool);
 
@@ -403,7 +417,7 @@ impl Queue {
             name: name.clone(),
             queue: TimeQ::new(),
             last_change: Instant::now(),
-            queue_config,
+            queue_config: ConfigHandle::new(queue_config),
             delayed_gauge,
             activity,
             rr,
@@ -444,12 +458,15 @@ impl Queue {
     ) -> anyhow::Result<Option<Message>> {
         let id = *msg.id();
         msg.increment_num_attempts();
-        let delay = self.queue_config.delay_for_attempt(msg.get_num_attempts());
+        let delay = self
+            .queue_config
+            .borrow()
+            .delay_for_attempt(msg.get_num_attempts());
         let jitter = (rand::random::<f32>() * 60.) - 30.0;
         let delay = chrono::Duration::seconds(delay.num_seconds() + jitter as i64);
 
         let now = Utc::now();
-        let max_age = self.queue_config.get_max_age();
+        let max_age = self.queue_config.borrow().get_max_age();
         let age = msg.age(now);
         let delayed_age = age + delay;
         if delayed_age > max_age {
@@ -469,7 +486,7 @@ impl Queue {
                     content: format!("Next delivery time {delayed_age} > {max_age}"),
                     command: None,
                 },
-                egress_pool: self.queue_config.egress_pool.as_deref(),
+                egress_pool: self.queue_config.borrow().egress_pool.as_deref(),
                 egress_source: None,
                 relay_disposition: None,
                 delivery_protocol: None,
@@ -564,7 +581,13 @@ impl Queue {
     #[instrument(skip(self, msg))]
     async fn insert_ready(&mut self, msg: Message) -> anyhow::Result<()> {
         tracing::trace!("insert_ready {}", msg.id());
-        match &self.queue_config.protocol {
+
+        let protocol = {
+            let config = self.queue_config.borrow();
+            config.protocol.clone()
+        };
+
+        match protocol {
             DeliveryProto::Smtp { .. } | DeliveryProto::Lua { .. } => {
                 // rr_attempts is a bit gross; ideally rr.next would know how
                 // to inspect the egress_path.suspended configuration and reflect
@@ -816,7 +839,7 @@ impl Queue {
         }
     }
 
-    pub fn get_config(&self) -> &QueueConfig {
+    pub fn get_config(&self) -> &ConfigHandle<QueueConfig> {
         &self.queue_config
     }
 }
@@ -880,6 +903,7 @@ async fn maintain_named_queue(queue: &QueueHandle) -> anyhow::Result<()> {
     let mut sleep_duration = Duration::from_secs(60);
     let mut shutdown = ShutdownSubcription::get();
     let mut memory = kumo_server_memory::subscribe_to_memory_status_changes();
+    let mut last_config_refresh = Instant::now();
 
     loop {
         tokio::select! {
@@ -920,6 +944,17 @@ async fn maintain_named_queue(queue: &QueueHandle) -> anyhow::Result<()> {
                 continue;
             }
 
+            if last_config_refresh.elapsed() >= Duration::from_secs(60) {
+                last_config_refresh = Instant::now();
+                if let Ok(mut config) = load_config().await {
+                    if let Ok(queue_config) =
+                        Queue::call_get_queue_config(&q.name, &mut config).await
+                    {
+                        q.queue_config.update(queue_config);
+                    }
+                }
+            }
+
             match q.queue.pop() {
                 PopResult::Items(messages) => {
                     q.delayed_gauge.sub(messages.len() as i64);

crates/kumod/src/ready_queue.rs

@@ -67,7 +67,7 @@ impl ReadyQueueManager {
 
     pub async fn compute_queue_name(
         queue_name: &str,
-        queue_config: &QueueConfig,
+        queue_config: &ConfigHandle<QueueConfig>,
         egress_source: &str,
     ) -> anyhow::Result<ReadyQueueName> {
         let components = QueueNameComponents::parse(queue_name);
@@ -85,7 +85,7 @@ impl ReadyQueueManager {
         // site name, we simply use the domain; we do not include the campaign
         // or tenant because those have no bearing from the perspective of
         // the recipient.
-        let site_name = match &queue_config.protocol {
+        let site_name = match &queue_config.borrow().protocol {
             DeliveryProto::Smtp { smtp } => {
                 if smtp.mx_list.is_empty() {
                     mx.replace(MailExchanger::resolve(routing_domain).await?);
@@ -102,7 +102,7 @@ impl ReadyQueueManager {
         // tenant or campaign to vary the protocol.
         let name = format!(
             "{egress_source}->{site_name}@{}",
-            queue_config.protocol.ready_queue_name()
+            queue_config.borrow().protocol.ready_queue_name()
         );
 
         Ok(ReadyQueueName {
@@ -114,7 +114,7 @@ impl ReadyQueueManager {
 
     async fn compute_config(
         queue_name: &str,
-        queue_config: &QueueConfig,
+        queue_config: &ConfigHandle<QueueConfig>,
         egress_source: &str,
     ) -> anyhow::Result<ReadyQueueConfig> {
         let ReadyQueueName {
@@ -159,7 +159,7 @@ impl ReadyQueueManager {
 
     pub async fn resolve_by_queue_name(
         queue_name: &str,
-        queue_config: &QueueConfig,
+        queue_config: &ConfigHandle<QueueConfig>,
         egress_source: &str,
         egress_pool: &str,
     ) -> anyhow::Result<ReadyQueueHandle> {
@@ -184,7 +184,7 @@ impl ReadyQueueManager {
                 move || Ok(async move { Self::maintainer_task(name).await })
             })
             .expect("failed to spawn maintainer");
-            let proto = queue_config.protocol.metrics_protocol_name();
+            let proto = queue_config.borrow().protocol.metrics_protocol_name();
             let service = format!("{proto}:{name}");
             let metrics = DeliveryMetrics::new(&service, &proto);
             let ready = Arc::new(StdMutex::new(VecDeque::new()));
@@ -304,7 +304,7 @@ pub struct ReadyQueue {
     activity: Activity,
     consecutive_connection_failures: Arc<AtomicUsize>,
     path_config: ConfigHandle<EgressPathConfig>,
-    queue_config: QueueConfig,
+    queue_config: ConfigHandle<QueueConfig>,
     egress_pool: String,
     egress_source: EgressSource,
 }
@@ -551,7 +551,7 @@ impl Dispatcher {
         mx: Option<Arc<MailExchanger>>,
         ready: Arc<StdMutex<VecDeque<Message>>>,
         notify: Arc<Notify>,
-        queue_config: QueueConfig,
+        queue_config: ConfigHandle<QueueConfig>,
         path_config: ConfigHandle<EgressPathConfig>,
         metrics: DeliveryMetrics,
         consecutive_connection_failures: Arc<AtomicUsize>,
@@ -561,7 +561,7 @@ impl Dispatcher {
     ) -> anyhow::Result<()> {
         let activity = Activity::get(format!("ready_queue Dispatcher {name}"))?;
 
-        let delivery_protocol = match &queue_config.protocol {
+        let delivery_protocol = match &queue_config.borrow().protocol {
             DeliveryProto::Smtp { .. } => "ESMTP".to_string(),
             DeliveryProto::Lua { .. } => "Lua".to_string(),
             DeliveryProto::Maildir { .. } => "Maildir".to_string(),
@@ -585,7 +585,7 @@ impl Dispatcher {
             lease,
         };
 
-        let mut queue_dispatcher: Box<dyn QueueDispatcher> = match &queue_config.protocol {
+        let mut queue_dispatcher: Box<dyn QueueDispatcher> = match &queue_config.borrow().protocol {
             DeliveryProto::Smtp { smtp } => {
                 match SmtpDispatcher::init(&mut dispatcher, smtp).await? {
                     Some(disp) => Box::new(disp),

crates/kumod/src/spool.rs

@@ -238,12 +238,15 @@ impl SpoolManager {
                                     let mut queue = queue.lock().await;
 
                                     let queue_config = queue.get_config();
-                                    let max_age = queue_config.get_max_age();
+                                    let max_age = queue_config.borrow().get_max_age();
                                     let age = msg.age(now);
-                                    let num_attempts = queue_config.infer_num_attempts(age);
+                                    let num_attempts =
+                                        queue_config.borrow().infer_num_attempts(age);
                                     msg.set_num_attempts(num_attempts);
 
-                                    match queue_config.compute_delay_based_on_age(num_attempts, age)
+                                    match queue_config
+                                        .borrow()
+                                        .compute_delay_based_on_age(num_attempts, age)
                                     {
                                         None => {
                                             tracing::debug!("expiring {id} {age} > {max_age}");

docs/reference/events/get_queue_config.md

@@ -1,4 +1,4 @@
-# `kumo.on('get_queue_config', function(domain, tenant, campaign))`
+# `kumo.on('get_queue_config', function(domain, tenant, campaign, routing_domain))`
 
 !!! note
     This event handler is in flux and may change significantly
@@ -6,12 +6,11 @@
 Not the final form of this API, but this is currently how
 we retrieve configuration used for managing a queue.
 
-The parameters correspond to the `domain`, `tenant` and `campaign`
+The parameters correspond to the `domain`, `tenant`, `campaign` and `routing_domain`
 fields from the *scheduled queue* name, as discussed in [Queues](../queues.md).
-The `routing_domain` is not passed to this event.
 
 ```lua
-kumo.on('get_queue_config', function(domain_name, tenant, campaign)
+kumo.on('get_queue_config', function(domain_name, tenant, campaign, routing_domain)
   return kumo.make_queue_config {
     max_retry_interval = '20 minutes',
   }