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 = {
+    // inbuxa: upstream didn't start the task manager on a node whose role
-        let server = inner.build_server();
+    // had no task types at boot, so adding one later did nothing until a
-        let roles = &server.core.network.roles;
+    // 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
-        // inbuxa: outbound_mta too, which now governs report tasks
+    // settings reload.
-        if !roles.account_maintenance
+    let is_clustered = inner.build_server().core.storage.coordinator.is_enabled();
-            && !roles.store_maintenance
-            && !roles.search_indexing
-            && !roles.spam_training
-            && !roles.outbound_mta
-            && !roles.task_manager
-        {
-            return;
-        }
-
-        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;