diff --git a/crates/common/src/cache/reload.rs b/crates/common/src/cache/reload.rs index abf7aa2..e8fd7b7 100644 --- a/crates/common/src/cache/reload.rs +++ b/crates/common/src/cache/reload.rs @@ -155,6 +155,12 @@ impl Server { .await .ok(); + // inbuxa: the task manager reads the node's role on + // every scan; scan now, so a role that gained task + // types starts claiming them without waiting out the + // refresh interval + self.inner.ipc.task_tx.notify_one(); + self.record_build_errors(&bootstrap.errors); return Ok(ReloadResult { diff --git a/crates/services/src/task_manager/manager.rs b/crates/services/src/task_manager/manager.rs index e7d0d1d..16142a2 100644 --- a/crates/services/src/task_manager/manager.rs +++ b/crates/services/src/task_manager/manager.rs @@ -55,23 +55,12 @@ const PERPETUAL_RETRY_MIN_DELAY: u64 = 3600; const PERPETUAL_RETRY_MAX_DELAY: u64 = 21600; pub fn spawn_task_manager(inner: Arc) { - let is_clustered = { - let server = inner.build_server(); - let roles = &server.core.network.roles; - - // inbuxa: outbound_mta too, which now governs report tasks - if !roles.account_maintenance - && !roles.store_maintenance - && !roles.search_indexing - && !roles.spam_training - && !roles.outbound_mta - && !roles.task_manager - { - return; - } - - server.core.storage.coordinator.is_enabled() - }; + // inbuxa: upstream didn't start the task manager on a node whose role + // had no task types at boot, so adding one later did nothing until a + // restart. It now always runs and reads the role on every scan and + // before every job (task_enabled), so a role change applies at the next + // settings reload. + let is_clustered = inner.build_server().core.storage.coordinator.is_enabled(); trc::event!(TaskManager(TaskManagerEvent::ManagerStarted)); @@ -151,20 +140,23 @@ pub fn spawn_task_manager(inner: Arc) { let server = inner.build_server(); let batch_size = server.core.email.index_batch_size; let mut batch = Vec::with_capacity(batch_size); - if let Some(task) = fetch_task(&server, job).await { + if let Some(task) = fetch_enabled_task(&server, job).await { batch.push(task); } while batch.len() < batch_size { match rx.try_recv() { Ok(job) => { - if let Some(task) = fetch_task(&server, job).await { + if let Some(task) = fetch_enabled_task(&server, job).await { batch.push(task); } } Err(_) => break, } } + if batch.is_empty() { + continue; + } // Dispatch. inbuxa: on a task of its own, so a panic // releases the batch's locks and leaves this worker @@ -205,7 +197,8 @@ pub fn spawn_task_manager(inner: Arc) { let server = inner.build_server(); let mut refresh_queue = false; - if let Some(TaskDetails { task, info }) = fetch_task(&server, job).await { + if let Some(TaskDetails { task, info }) = fetch_enabled_task(&server, job).await + { // inbuxa: on a task of its own, as above let run = { let server = server.clone(); @@ -274,6 +267,17 @@ impl TaskQueueManager for Server { if task_locks.is_stopping() { return Duration::from_secs(QUEUE_REFRESH_INTERVAL); } + // inbuxa: with no task type enabled by this node's role there is + // nothing to claim; a settings reload wakes the manager when that + // changes + let roles = &self.core.network.roles; + if !(0..TaskType::COUNT as u16) + .filter_map(TaskType::from_id) + .any(|task_type| task_enabled(roles, task_type)) + { + ipc.locked.clear(); + return Duration::from_secs(QUEUE_REFRESH_INTERVAL); + } let lock_expiry = task_locks.expiry(); let now_timestamp = now(); let from_key = ValueKey:: { @@ -296,7 +300,6 @@ impl TaskQueueManager for Server { let mut tasks = Vec::new(); let now = Instant::now(); let mut next_event = None; - let roles = &self.core.network.roles; ipc.revision += 1; let _ = self .store() @@ -544,6 +547,25 @@ async fn run_task( } } +/// inbuxa: reads a claimed task when this node's role still allows its type. +/// The role may have changed since the task was claimed (a settings reload in +/// between); the claim is then handed back at once for a node that may run +/// it, rather than held until the lease runs out. +async fn fetch_enabled_task(server: &Server, job: TaskJob) -> Option { + if task_enabled(&server.core.network.roles, job.typ) { + fetch_task(server, job).await + } else { + trc::event!( + TaskManager(TaskManagerEvent::TaskIgnored), + Id = job.id, + Details = job.typ.as_str(), + Reason = "Task type was disabled by cluster roles after it was claimed.", + ); + server.remove_index_lock(job.id).await; + None + } +} + /// Reads a claimed task. When it is gone or can't be read, the claim is /// released: inbuxa: holding it would block the task, everywhere, until /// the lock expired. diff --git a/crates/smtp/src/lib.rs b/crates/smtp/src/lib.rs index 8f8403a..c2e0238 100644 --- a/crates/smtp/src/lib.rs +++ b/crates/smtp/src/lib.rs @@ -44,7 +44,14 @@ impl StartQueueManager for BootManager { impl SpawnQueueManager for IpcReceivers { fn spawn_queue_manager(&mut self, inner: Arc) { let core = inner.shared_core.load(); - if !core.storage.registry.is_recovery_mode() && core.network.roles.outbound_mta { + // inbuxa: upstream started these only when the node's role included + // outboundMta at boot, so turning the role on later did nothing and + // turning it off left them delivering until a restart. They now run + // on every node and follow the role live (see Queue::start and the + // report scheduler). This also drains the queue channel on nodes + // without the role, where every queued message's refresh used to sit + // in a channel nobody read until it filled and queueing blocked. + if !core.storage.registry.is_recovery_mode() { // Spawn queue manager self.queue_rx.take().unwrap().spawn(inner.clone()); diff --git a/crates/smtp/src/queue/manager.rs b/crates/smtp/src/queue/manager.rs index 0719360..18c87e0 100644 --- a/crates/smtp/src/queue/manager.rs +++ b/crates/smtp/src/queue/manager.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use super::{Message, QueueId, Status, spool::SmtpSpool}; @@ -39,6 +41,9 @@ pub struct Queue { pub urgent_refresh: bool, pub last_scan: Instant, pub last_full_scan: Instant, + /// inbuxa: whether this node's role included outboundMta when last + /// checked (None before the first check) + pub role_enabled: Option, } #[derive(Debug)] @@ -67,6 +72,9 @@ impl SpawnQueue for mpsc::Receiver { const BACK_PRESSURE_WARN_INTERVAL: Duration = Duration::from_secs(60); const MIN_SCAN_INTERVAL: Duration = Duration::from_millis(100); const FULL_SCAN_INTERVAL: Duration = Duration::from_secs(QUEUE_REFRESH / 2); +/// inbuxa: how often a node without the outbound MTA role looks at its role +/// again when nothing else wakes it (a settings reload does) +const ROLE_RECHECK_INTERVAL: Duration = Duration::from_secs(30); impl Queue { pub fn new(core: Arc, rx: mpsc::Receiver) -> Self { @@ -87,6 +95,7 @@ impl Queue { urgent_refresh: false, last_scan: now.checked_sub(MIN_SCAN_INTERVAL).unwrap_or(now), last_full_scan: now, + role_enabled: None, } } @@ -123,6 +132,27 @@ impl Queue { continue; } + // inbuxa: follow the node's role live. Without outboundMta the + // queue claims nothing new; deliveries already running finish + // and report back as usual, releasing their locks. When the role + // comes back, the whole queue is scanned at once. + let role_enabled = self.core.shared_core.load().network.roles.outbound_mta; + if self.role_enabled.replace(role_enabled) == Some(false) && role_enabled { + trc::event!( + Queue(trc::QueueEvent::Started), + Details = "This node's cluster role now includes outboundMta", + ); + self.scan_from = 0; + self.pending_refresh = true; + self.urgent_refresh = true; + } + if !role_enabled { + self.pending_refresh = false; + self.urgent_refresh = false; + self.next_refresh = Instant::now() + ROLE_RECHECK_INTERVAL; + continue; + } + self.pending_refresh |= refresh_queue; if !self.pending_refresh && self.next_refresh > Instant::now() { continue; diff --git a/crates/smtp/src/reporting/scheduler.rs b/crates/smtp/src/reporting/scheduler.rs index b28b7c2..4b08f7b 100644 --- a/crates/smtp/src/reporting/scheduler.rs +++ b/crates/smtp/src/reporting/scheduler.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use super::{dmarc::DmarcReporting, tls::TlsReporting}; @@ -18,6 +20,14 @@ impl SpawnReport for mpsc::Receiver { tokio::spawn(async move { while let Some(event) = self.recv().await { let server = inner.build_server(); + // inbuxa: reports are the outbound MTA's business, as at + // boot, but the role is read per event so a change applies + // without a restart. Events that arrive while the role is + // off are dropped, as they were on a node started without it + if !matches!(event, ReportingEvent::Stop) && !server.core.network.roles.outbound_mta + { + continue; + } match event { ReportingEvent::Dmarc(event) => server.schedule_dmarc(event).await, ReportingEvent::Tls(event) => server.schedule_tls(event).await, diff --git a/tests/src/cluster/live_roles.rs b/tests/src/cluster/live_roles.rs new file mode 100644 index 0000000..67a923d --- /dev/null +++ b/tests/src/cluster/live_roles.rs @@ -0,0 +1,297 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! A node follows edits to its cluster role without a restart: outbound +//! delivery and report tasks start when the role gains outboundMta and stop +//! when it loses it. Upstream decided at boot whether the queue, report and +//! task managers ran at all. Needs a store the seed and the node can share +//! (STORE=PostgreSql or MySql). + +use crate::utils::server::{TestServer, TestServerBuilder}; +use common::Server; +use registry::{ + schema::{ + enums::ClusterTaskType, + prelude::{Object, ObjectType}, + structs::{ + ClusterListenerGroup, ClusterRole, ClusterTaskGroup, ClusterTaskGroupProperties, Task, + TaskStatus, TaskTlsReport, + }, + }, + types::{id::ObjectId, map::Map}, +}; +use smtp::{ + queue::{Message, Status}, + reporting::send::MtaReportSend, +}; +use std::time::{Duration, Instant}; +use store::{ + Deserialize, IterateParams, ValueKey, + registry::write::{RegistryWrite, RegistryWriteResult}, + write::{AlignedBytes, Archive, BatchBuilder, QueueClass, TaskQueueClass, ValueClass}, +}; +use types::id::Id; +use utils::snowflake::SnowflakeIdGenerator; + +const BUSY_ROLE: &str = "live_role_busy"; +const IDLE_ROLE: &str = "live_role_idle"; +const WITH: &[ClusterTaskType] = &[ + ClusterTaskType::PushNotifications, + ClusterTaskType::OutboundMta, +]; +const WITHOUT: &[ClusterTaskType] = &[ClusterTaskType::PushNotifications]; +const RCPT_DOMAIN: &str = "live-role.invalid"; + +#[tokio::test(flavor = "multi_thread")] +pub async fn live_role_tests() { + if matches!( + std::env::var("STORE").as_deref(), + Ok("RocksDb" | "Sqlite") | Err(_) + ) { + println!("Skipping live role tests: they need a store the nodes can share."); + return; + } + println!( + "Running live role tests on {}...", + std::env::var("STORE").unwrap_or_default() + ); + + // Two roles: one with outboundMta, one with no task type at all + let seed = TestServerBuilder::new("live_roles_seed").await; + let busy_id = seed.insert_object(role(BUSY_ROLE, WITH)).await; + let idle_id = seed.insert_object(role(IDLE_ROLE, WITHOUT)).await; + let seed = seed.disable_services().build().await; + let registry = seed.server.clone(); + + // 1. The rehearsal case: a node started with outboundMta has it taken + // away. Upstream kept delivering, report messages included, until a + // restart. + let node = start_node("live_roles_busy", BUSY_ROLE).await; + let server = node.server.clone(); + assert!(server.core.network.roles.outbound_mta); + let (msg, task) = queue_work(&server, "busy-before").await; + assert_runs(&server, &msg, task).await; + + set_role(®istry, &server, busy_id, role(BUSY_ROLE, WITHOUT)).await; + let (msg, task) = queue_work(&server, "busy-off").await; + assert_idle(&server, &msg, task).await; + + // Given back, it takes up the work left waiting + set_role(®istry, &server, busy_id, role(BUSY_ROLE, WITH)).await; + assert_runs(&server, &msg, task).await; + + // Off again, so it leaves the next node's work alone + set_role(®istry, &server, busy_id, role(BUSY_ROLE, WITHOUT)).await; + + // 2. A node started with no task type at all gains outboundMta. + // Upstream never started its queue, report or task manager, so the + // role did nothing until a restart. + let node2 = start_node("live_roles_idle", IDLE_ROLE).await; + let server2 = node2.server.clone(); + assert!(!server2.core.network.roles.outbound_mta); + let (msg, task) = queue_work(&server2, "idle-off").await; + assert_idle(&server2, &msg, task).await; + set_role(®istry, &server2, idle_id, role(IDLE_ROLE, WITH)).await; + assert_runs(&server2, &msg, task).await; + set_role(®istry, &server2, idle_id, role(IDLE_ROLE, WITHOUT)).await; + + if seed.is_reset() { + seed.temp_dir.delete(); + node.temp_dir.delete(); + node2.temp_dir.delete(); + } +} + +async fn start_node(name: &str, role: &str) -> TestServer { + TestServerBuilder::new_with_role( + name, + format!("{name}.example.com").replace('_', "-"), + Some(role.into()), + false, + ) + .await + .build_with_opts(false) + .await +} + +/// Neither the message nor the report task is touched. +async fn assert_idle(server: &Server, msg: &str, task: u64) { + tokio::time::sleep(Duration::from_secs(4)).await; + server.notify_task_queue(); + tokio::time::sleep(Duration::from_secs(1)).await; + assert!( + !attempted(server, msg).await, + "delivery attempted without outboundMta" + ); + assert!( + is_pending(server, task).await, + "report task claimed without outboundMta" + ); +} + +/// Delivery of the message is attempted and the report task runs. +async fn assert_runs(server: &Server, msg: &str, task: u64) { + wait_for(Duration::from_secs(20), "message delivery attempt", || { + attempted(server, msg) + }) + .await; + wait_for(Duration::from_secs(20), "report task to run", || async { + !is_pending(server, task).await + }) + .await; +} + +fn role(name: &str, tasks: &[ClusterTaskType]) -> ClusterRole { + ClusterRole { + name: name.into(), + description: None, + listeners: ClusterListenerGroup::EnableAll, + tasks: ClusterTaskGroup::EnableSome(ClusterTaskGroupProperties { + task_types: Map::new(tasks.to_vec()), + }), + } +} + +/// Stores a new version of a role and reloads the node's settings, as a +/// JMAP write to the role does. +async fn set_role(registry: &Server, node: &Server, id: Id, new: ClusterRole) { + let enabled = matches!(&new.tasks, ClusterTaskGroup::EnableSome(group) + if group.task_types.iter().any(|t| *t == ClusterTaskType::OutboundMta)); + let old = registry + .registry() + .get(ObjectId::new(ObjectType::ClusterRole, id)) + .await + .unwrap() + .expect("role not found"); + let new = Object::from(new); + let result = registry + .registry() + .write(RegistryWrite::update(id, &new, &old)) + .await + .unwrap(); + assert!( + matches!(result, RegistryWriteResult::Success(_)), + "role update refused" + ); + assert_eq!( + node.reload_after_write(ObjectType::ClusterRole).await, + Some(Ok(())) + ); + assert_eq!( + node.inner.shared_core.load().network.roles.outbound_mta, + enabled + ); +} + +/// Queues a message to an unreachable domain and schedules a TLS report +/// task, both due now. Returns the recipient's local part and the task id. +async fn queue_work(server: &Server, name: &str) -> (String, u64) { + let local = format!("{name}-{}", SnowflakeIdGenerator::global_id().unwrap()); + let rcpt = format!("{local}@{RCPT_DOMAIN}"); + server + .send_autogenerated( + "postmaster@example.com", + [rcpt.as_str()].into_iter(), + format!( + "From: postmaster@example.com\r\nTo: {rcpt}\r\n\ + Subject: live role test\r\n\r\nTest\r\n" + ) + .into_bytes(), + None, + 0, + ) + .await; + assert!( + queued_recipient(server, &rcpt).await.is_some(), + "message to {rcpt} was not queued" + ); + + let task = SnowflakeIdGenerator::global_id().unwrap(); + let mut batch = BatchBuilder::new(); + batch.schedule_task_with_id( + task, + Task::TlsReport(TaskTlsReport { + report_id: u64::MAX.into(), + status: TaskStatus::now(), + }), + ); + server.store().write(batch.build_all()).await.unwrap(); + server.notify_task_queue(); + + (rcpt, task) +} + +/// Whether delivery to `rcpt` was tried: the message is gone, or its +/// recipient is no longer scheduled or has a retry count. +async fn attempted(server: &Server, rcpt: &str) -> bool { + match queued_recipient(server, rcpt).await { + None => true, + Some((status_scheduled, retries)) => !status_scheduled || retries > 0, + } +} + +/// The queued recipient `rcpt`: whether it is still scheduled, and how many +/// times delivery was retried. +async fn queued_recipient(server: &Server, rcpt: &str) -> Option<(bool, u32)> { + let mut found = None; + server + .store() + .iterate( + IterateParams::new( + ValueKey::from(ValueClass::Queue(QueueClass::Message(0))), + ValueKey::from(ValueClass::Queue(QueueClass::Message(u64::MAX))), + ), + |_, value| { + let message = as Deserialize>::deserialize(value)? + .deserialize::()?; + if let Some(recipient) = message + .recipients + .iter() + .find(|recipient| recipient.address.as_ref() == rcpt) + { + found = Some(( + matches!(recipient.status, Status::Scheduled), + recipient.retry.inner, + )); + return Ok(false); + } + Ok(true) + }, + ) + .await + .unwrap(); + found +} + +async fn is_pending(server: &Server, id: u64) -> bool { + matches!( + server + .store() + .get_value::(ValueKey::from(ValueClass::TaskQueue( + TaskQueueClass::Task { id }, + ))) + .await + .unwrap() + .map(|task| task.status().clone()), + Some(TaskStatus::Pending(_)) + ) +} + +async fn wait_for(within: Duration, what: &str, mut check: F) +where + F: FnMut() -> Fut, + Fut: Future, +{ + let started = Instant::now(); + while !check().await { + assert!( + started.elapsed() < within, + "still waiting for the {what} after {:?}", + started.elapsed() + ); + tokio::time::sleep(Duration::from_millis(250)).await; + } +} diff --git a/tests/src/cluster/mod.rs b/tests/src/cluster/mod.rs index 3efa3f6..ae79136 100644 --- a/tests/src/cluster/mod.rs +++ b/tests/src/cluster/mod.rs @@ -7,6 +7,7 @@ */ pub mod broadcast; +pub mod live_roles; // inbuxa: role edits apply without a restart #[cfg(feature = "nats")] pub mod coordinator; // inbuxa: coordinator reconnects pub mod stress;