diff --git a/crates/kumod/src/egress_source.rs b/crates/kumod/src/egress_source.rs index 0db94e1e..619ba563 100644 --- a/crates/kumod/src/egress_source.rs +++ b/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, + ) -> RoundRobinResult { if self.entries.is_empty() { return RoundRobinResult::NoSources; } diff --git a/crates/kumod/src/queue.rs b/crates/kumod/src/queue.rs index 3d094fff..ec380ea0 100644 --- a/crates/kumod/src/queue.rs +++ b/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, last_change: Instant, - queue_config: QueueConfig, + queue_config: ConfigHandle, delayed_gauge: IntGauge, activity: Activity, rr: EgressPoolRoundRobin, } impl Queue { - pub async fn new(name: String) -> anyhow::Result { - let mut config = load_config().await?; - + async fn call_get_queue_config( + name: &str, + config: &mut LuaConfig, + ) -> anyhow::Result { 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 { + 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> { 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::() * 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 { &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); diff --git a/crates/kumod/src/ready_queue.rs b/crates/kumod/src/ready_queue.rs index 4326643c..492f6843 100644 --- a/crates/kumod/src/ready_queue.rs +++ b/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, egress_source: &str, ) -> anyhow::Result { 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, egress_source: &str, ) -> anyhow::Result { let ReadyQueueName { @@ -159,7 +159,7 @@ impl ReadyQueueManager { pub async fn resolve_by_queue_name( queue_name: &str, - queue_config: &QueueConfig, + queue_config: &ConfigHandle, egress_source: &str, egress_pool: &str, ) -> anyhow::Result { @@ -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, path_config: ConfigHandle, - queue_config: QueueConfig, + queue_config: ConfigHandle, egress_pool: String, egress_source: EgressSource, } @@ -551,7 +551,7 @@ impl Dispatcher { mx: Option>, ready: Arc>>, notify: Arc, - queue_config: QueueConfig, + queue_config: ConfigHandle, path_config: ConfigHandle, metrics: DeliveryMetrics, consecutive_connection_failures: Arc, @@ -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 = match &queue_config.protocol { + let mut queue_dispatcher: Box = match &queue_config.borrow().protocol { DeliveryProto::Smtp { smtp } => { match SmtpDispatcher::init(&mut dispatcher, smtp).await? { Some(disp) => Box::new(disp), diff --git a/crates/kumod/src/spool.rs b/crates/kumod/src/spool.rs index 47104084..eb10eacb 100644 --- a/crates/kumod/src/spool.rs +++ b/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}"); diff --git a/docs/reference/events/get_queue_config.md b/docs/reference/events/get_queue_config.md index 183f0c7c..af622a77 100644 --- a/docs/reference/events/get_queue_config.md +++ b/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', }