From 1543ea5a9e80e5e46578e0eca9616844435176b3 Mon Sep 17 00:00:00 2001 From: John Coffey Date: Thu, 24 Sep 2026 12:46:49 -0700 Subject: [PATCH] Task manager: every task type follows the node's cluster role A 3-node rehearsal found taskQueueProcessing didn't filter anything: roles.task_manager only decided whether the task manager started, and report, ACME, DKIM, DNS, calendar, thread-merge and restore tasks ran on any node with a task manager (manager.rs returned true for them). A node whose role left taskQueueProcessing off still ran them if it indexed or did maintenance. Every task type now answers to one ClusterTaskType (task_enabled): - 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 (no lock), so a node that may picks it up. The task manager also starts on a node whose only task role is outboundMta, so reports still run there. cluster::task_roles::task_role_tests (new, two task managers over one PostgreSQL store): node A (taskQueueProcessing only) runs a DNS task and leaves an unindex task and a TLS report pending; node B (searchIndexing and outboundMta) comes up and runs those two; a DNS task scheduled next stays pending on B and runs on A. On main node A runs the TLS report. --- crates/services/src/task_manager/manager.rs | 66 ++++-- tests/src/cluster/mod.rs | 1 + tests/src/cluster/task_roles.rs | 218 ++++++++++++++++++++ 3 files changed, 265 insertions(+), 20 deletions(-) create mode 100644 tests/src/cluster/task_roles.rs diff --git a/crates/services/src/task_manager/manager.rs b/crates/services/src/task_manager/manager.rs index 4d32b9a..1906e5b 100644 --- a/crates/services/src/task_manager/manager.rs +++ b/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) { 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, diff --git a/tests/src/cluster/mod.rs b/tests/src/cluster/mod.rs index 6d06c6f..3efa3f6 100644 --- a/tests/src/cluster/mod.rs +++ b/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 diff --git a/tests/src/cluster/task_roles.rs b/tests/src/cluster/task_roles.rs new file mode 100644 index 0000000..34a6f22 --- /dev/null +++ b/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() -> [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::(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; + } +}