Cluster role changes apply to delivery and tasks without a restart

In cluster rehearsal 3, turning outboundMta off on node1's role was
reported applied (x:settingsReload applied: true), yet node1 kept
delivering mail, a report message included, until it was restarted.
The queue and report managers were started at boot only when the
node's role included outboundMta (crates/smtp/src/lib.rs), and the task
manager only when the role had some task type (spawn_task_manager).
After that nothing looked at the role again: a queue manager that was
running kept claiming and delivering, and one that wasn't never
started.

They now start on every node (outside recovery mode) and follow the
role live:

- Queue manager: before each scan it reads the role from the running
  settings. Without outboundMta it claims nothing new; deliveries
  already running finish and report back as usual, which releases
  their locks. When the role comes back (a reload wakes the manager
  with ReloadSettings, and it looks again every 30 s regardless) it
  logs queue.started and scans the whole queue at once.
- Report scheduler: DMARC and TLS report events are handled only while
  the role has outboundMta, as at boot; events arriving without it are
  dropped, as they were on a node started without the role.
- Task manager: task_enabled already read the current role on every
  scan. It now also runs on nodes whose role has no task type (the
  scan returns at once until one is added), a job claimed before a
  role change is handed back at once rather than run or held until
  its lease lapses, and a settings reload wakes the manager so a role
  that gained task types starts claiming them straight away.

Starting the queue manager on every node also drains the queue channel
on nodes without outboundMta. Upstream left that channel unread, so
each message queued there parked a refresh in it, and by the code,
queueing would block once 1024 had piled up (not reproduced here).

A role object edit reaches the nodes that name that role in
INBUXA_ROLE. Moving a node to another role still means changing its
environment, and so a restart. Listener changes in a role still need a
restart too (listeners bind at boot); this change is about tasks and
delivery.

cluster::live_roles::live_role_tests (new; PostgreSQL, two nodes over
one store):
1. A node started with outboundMta delivers and runs a TLS report
   task; after its role loses outboundMta and the settings reload, a
   new message isn't attempted and a new report task stays pending;
   with the role back, both are taken up.
2. A node started with no task type at all gains outboundMta: a
   waiting message is attempted and a report task runs.
On main the test fails at step 1 ("delivery attempted without
outboundMta"); with step 1 bypassed, step 2 fails (nothing picked the
message up in 20 s).

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 {

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<Inner>) {
-    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<Inner>) {
                     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<Inner>) {
                     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::<ValueClass> {
@@ -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<TaskDetails> {
+    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.

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<Inner>) {
         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());
 

crates/smtp/src/queue/manager.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * 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<bool>,
 }
 
 #[derive(Debug)]
@@ -67,6 +72,9 @@ impl SpawnQueue for mpsc::Receiver<QueueEvent> {
 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<Inner>, rx: mpsc::Receiver<QueueEvent>) -> 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;

crates/smtp/src/reporting/scheduler.rs

@@ -2,6 +2,8 @@
  * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
  *
  * 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<ReportingEvent> {
         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,

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(&registry, &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(&registry, &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(&registry, &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(&registry, &server2, idle_id, role(IDLE_ROLE, WITH)).await;
+    assert_runs(&server2, &msg, task).await;
+    set_role(&registry, &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(
+            "[email protected]",
+            [rcpt.as_str()].into_iter(),
+            format!(
+                "From: [email protected]\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 = <Archive<AlignedBytes> as Deserialize>::deserialize(value)?
+                    .deserialize::<Message>()?;
+                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::<Task>(ValueKey::from(ValueClass::TaskQueue(
+                TaskQueueClass::Task { id },
+            )))
+            .await
+            .unwrap()
+            .map(|task| task.status().clone()),
+        Some(TaskStatus::Pending(_))
+    )
+}
+
+async fn wait_for<F, Fut>(within: Duration, what: &str, mut check: F)
+where
+    F: FnMut() -> Fut,
+    Fut: Future<Output = bool>,
+{
+    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;
+    }
+}

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;