crates/services/src/task_manager/manager.rs

@@ -24,6 +24,7 @@ use crate::task_manager::{
     TaskJob, TaskManagerIpc, TaskResult,
 };
 use common::BuildServer;
+use common::config::network::ClusterRoles;
 use common::config::server::{DEFAULT_TLS_TIMEOUT, ServerProtocol};
 use common::network::limiter::ConcurrencyLimiter;
 use common::network::{ServerInstance, TcpAcceptor};
@@ -58,10 +59,12 @@ pub fn spawn_task_manager(inner: Arc<Inner>) {
         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;
@@ -289,26 +292,7 @@ impl TaskQueueManager for Server {
                                     .caused_by(trc::location!())
                                     .ctx(trc::Key::Value, value)
                             })?;
-                            let enabled = match task_type {
-                                TaskType::IndexDocument
-                                | TaskType::UnindexDocument
-                                | TaskType::IndexTrace => roles.search_indexing,
-                                TaskType::AccountMaintenance
-                                | TaskType::TenantMaintenance
-                                | TaskType::DestroyAccount => roles.account_maintenance,
-                                TaskType::StoreMaintenance => roles.store_maintenance,
-                                TaskType::SpamFilterMaintenance => roles.spam_training,
-                                TaskType::CalendarAlarmEmail
-                                | TaskType::CalendarAlarmNotification
-                                | TaskType::CalendarItipMessage
-                                | TaskType::MergeThreads
-                                | TaskType::DmarcReport
-                                | TaskType::TlsReport
-                                | TaskType::RestoreArchivedItem
-                                | TaskType::AcmeRenewal
-                                | TaskType::DkimManagement
-                                | TaskType::DnsManagement => true,
-                            };
+                            let enabled = task_enabled(roles, task_type);
 
                             if !enabled {
                                 trc::event!(
@@ -437,6 +421,48 @@ impl TaskQueueManager for Server {
     }
 }
 
+/// inbuxa: whether this node's cluster role lets it run a task type. Upstream
+/// checked the dedicated roles (search indexing, account and store
+/// maintenance, spam training) and let every node with a task manager run
+/// the rest, whatever its taskQueueProcessing setting. Every task type now
+/// answers to one ClusterTaskType:
+///
+/// - IndexDocument, UnindexDocument, IndexTrace: searchIndexing
+/// - AccountMaintenance, TenantMaintenance, DestroyAccount: accountMaintenance
+/// - StoreMaintenance: storeMaintenance
+/// - SpamFilterMaintenance: spamClassifierTraining
+/// - DmarcReport, TlsReport: outboundMta. They build and send reports to
+///   other domains (TLS reports can go straight to an HTTPS endpoint), which
+///   is the outbound MTA's business.
+/// - CalendarAlarmEmail, CalendarAlarmNotification, CalendarItipMessage,
+///   MergeThreads, RestoreArchivedItem, AcmeRenewal, DkimManagement,
+///   DnsManagement: taskQueueProcessing, the role for queue tasks with no
+///   role of their own.
+///
+/// A node that may not run a task leaves it unclaimed, so a node that may
+/// picks it up.
+pub fn task_enabled(roles: &ClusterRoles, task_type: TaskType) -> bool {
+    match task_type {
+        TaskType::IndexDocument | TaskType::UnindexDocument | TaskType::IndexTrace => {
+            roles.search_indexing
+        }
+        TaskType::AccountMaintenance | TaskType::TenantMaintenance | TaskType::DestroyAccount => {
+            roles.account_maintenance
+        }
+        TaskType::StoreMaintenance => roles.store_maintenance,
+        TaskType::SpamFilterMaintenance => roles.spam_training,
+        TaskType::DmarcReport | TaskType::TlsReport => roles.outbound_mta,
+        TaskType::CalendarAlarmEmail
+        | TaskType::CalendarAlarmNotification
+        | TaskType::CalendarItipMessage
+        | TaskType::MergeThreads
+        | TaskType::RestoreArchivedItem
+        | TaskType::AcmeRenewal
+        | TaskType::DkimManagement
+        | TaskType::DnsManagement => roles.task_manager,
+    }
+}
+
 async fn run_task(
     server: &Server,
     task: &Task,

tests/src/cluster/mod.rs

@@ -10,3 +10,4 @@ pub mod broadcast;
 #[cfg(feature = "nats")]
 pub mod coordinator; // inbuxa: coordinator reconnects
 pub mod stress;
+pub mod task_roles; // inbuxa: task types follow cluster roles

tests/src/cluster/task_roles.rs

@@ -0,0 +1,218 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! Two task managers with different cluster roles over one shared store:
+//! each runs only the task types its role allows, and a task one node may
+//! not run is left for the node that may. Needs a store both nodes can open
+//! (STORE=PostgreSql or MySql).
+
+use crate::utils::server::TestServerBuilder;
+use common::Server;
+use registry::{
+    schema::{
+        enums::{ClusterTaskType, IndexDocumentType},
+        structs::{
+            ClusterListenerGroup, ClusterRole, ClusterTaskGroup, ClusterTaskGroupProperties, Task,
+            TaskDnsManagement, TaskIndexDocument, TaskStatus, TaskTlsReport,
+        },
+    },
+    types::map::Map,
+};
+use std::time::{Duration, Instant};
+use store::{
+    ValueKey,
+    write::{BatchBuilder, TaskQueueClass, ValueClass},
+};
+use utils::snowflake::SnowflakeIdGenerator;
+
+const QUEUE_ROLE: &str = "tasks_queue";
+const INDEX_MTA_ROLE: &str = "tasks_index_mta";
+
+#[tokio::test(flavor = "multi_thread")]
+pub async fn task_role_tests() {
+    if matches!(
+        std::env::var("STORE").as_deref(),
+        Ok("RocksDb" | "Sqlite") | Err(_)
+    ) {
+        println!("Skipping task role tests: they need a store both nodes can open.");
+        return;
+    }
+    println!(
+        "Running task role tests on {}...",
+        std::env::var("STORE").unwrap_or_default()
+    );
+
+    // The roles, stored by a node that runs no services of its own (a node
+    // looks its role up when it starts)
+    let seed = TestServerBuilder::new("task_roles_seed")
+        .await
+        .with_object(role(QUEUE_ROLE, &[ClusterTaskType::TaskQueueProcessing]))
+        .await
+        .with_object(role(
+            INDEX_MTA_ROLE,
+            &[
+                ClusterTaskType::SearchIndexing,
+                ClusterTaskType::OutboundMta,
+            ],
+        ))
+        .await
+        .disable_services()
+        .build()
+        .await;
+
+    // Node A runs queue tasks (taskQueueProcessing) only
+    let node_a = TestServerBuilder::new_with_role(
+        "task_roles_a",
+        "node-a.example.com".into(),
+        Some(QUEUE_ROLE.into()),
+        false,
+    )
+    .await
+    .build_with_opts(false)
+    .await;
+    let server_a = node_a.server.clone();
+    let roles = &server_a.core.network.roles;
+    assert!(roles.task_manager && !roles.search_indexing && !roles.outbound_mta);
+
+    // A DNS task (taskQueueProcessing), an unindex task (searchIndexing) and
+    // a TLS report (outboundMta), all due now
+    let [dns, unindex, report] = new_task_ids();
+    let mut batch = BatchBuilder::new();
+    batch
+        .schedule_task_with_id(
+            dns,
+            Task::DnsManagement(TaskDnsManagement {
+                status: TaskStatus::now(),
+                ..Default::default()
+            }),
+        )
+        .schedule_task_with_id(
+            unindex,
+            Task::UnindexDocument(TaskIndexDocument {
+                account_id: 0u32.into(),
+                document_id: u32::MAX.into(),
+                document_type: IndexDocumentType::File,
+                status: TaskStatus::now(),
+            }),
+        )
+        .schedule_task_with_id(
+            report,
+            Task::TlsReport(TaskTlsReport {
+                report_id: u64::MAX.into(),
+                status: TaskStatus::now(),
+            }),
+        );
+    server_a.store().write(batch.build_all()).await.unwrap();
+    server_a.notify_task_queue();
+
+    // Node A runs the DNS task and leaves the other two alone. Upstream ran
+    // the TLS report here too: report tasks ran on any node with a task
+    // manager.
+    wait_until_run(&server_a, &[dns], Duration::from_secs(20)).await;
+    tokio::time::sleep(Duration::from_secs(3)).await;
+    server_a.notify_task_queue();
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    assert!(
+        is_pending(&server_a, unindex).await,
+        "unindex ran on node A"
+    );
+    assert!(
+        is_pending(&server_a, report).await,
+        "TLS report ran on node A"
+    );
+
+    // Node B (search indexing and outbound MTA) comes up and picks up what
+    // node A left
+    let node_b = TestServerBuilder::new_with_role(
+        "task_roles_b",
+        "node-b.example.com".into(),
+        Some(INDEX_MTA_ROLE.into()),
+        false,
+    )
+    .await
+    .build_with_opts(false)
+    .await;
+    let server_b = node_b.server.clone();
+    let roles = &server_b.core.network.roles;
+    assert!(!roles.task_manager && roles.search_indexing && roles.outbound_mta);
+    server_b.notify_task_queue();
+    wait_until_run(&server_b, &[unindex, report], Duration::from_secs(20)).await;
+
+    // A queue task scheduled now still runs, on node A: node B may not
+    // claim it
+    let [dns] = new_task_ids();
+    let mut batch = BatchBuilder::new();
+    batch.schedule_task_with_id(
+        dns,
+        Task::DnsManagement(TaskDnsManagement {
+            status: TaskStatus::now(),
+            ..Default::default()
+        }),
+    );
+    server_b.store().write(batch.build_all()).await.unwrap();
+    server_b.notify_task_queue();
+    tokio::time::sleep(Duration::from_secs(3)).await;
+    assert!(is_pending(&server_b, dns).await, "DNS task ran on node B");
+    server_a.notify_task_queue();
+    wait_until_run(&server_a, &[dns], Duration::from_secs(20)).await;
+
+    if seed.is_reset() {
+        seed.temp_dir.delete();
+        node_a.temp_dir.delete();
+        node_b.temp_dir.delete();
+    }
+}
+
+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()),
+        }),
+    }
+}
+
+fn new_task_ids<const N: usize>() -> [u64; N] {
+    std::array::from_fn(|_| SnowflakeIdGenerator::global_id().unwrap())
+}
+
+/// Still due and never run: present, and pending.
+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_until_run(server: &Server, ids: &[u64], within: Duration) {
+    let started = Instant::now();
+    loop {
+        let mut left = 0;
+        for id in ids {
+            if is_pending(server, *id).await {
+                left += 1;
+            }
+        }
+        if left == 0 {
+            return;
+        }
+        assert!(
+            started.elapsed() < within,
+            "{left} task(s) still pending after {:?}",
+            started.elapsed()
+        );
+        tokio::time::sleep(Duration::from_millis(250)).await;
+    }
+}