diff --git a/crates/smtp/src/lib.rs b/crates/smtp/src/lib.rs index c2e0238..ec9b94c 100644 --- a/crates/smtp/src/lib.rs +++ b/crates/smtp/src/lib.rs @@ -47,8 +47,10 @@ impl SpawnQueueManager for IpcReceivers { // 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 + // on every node: the queue follows the role live (see Queue::start), + // and the report scheduler records DMARC and TLS results on every + // node, whatever its role (see reporting/scheduler.rs). 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() { diff --git a/crates/smtp/src/reporting/dmarc.rs b/crates/smtp/src/reporting/dmarc.rs index 57f1272..c24a03c 100644 --- a/crates/smtp/src/reporting/dmarc.rs +++ b/crates/smtp/src/reporting/dmarc.rs @@ -2,9 +2,12 @@ * 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::AggregateTimestamp; +use super::shared::{MAX_WRITE_RETRIES, Revisioned, write_retry_pause}; use crate::{ core::Session, queue::RecipientDomain, @@ -349,29 +352,43 @@ impl DmarcReporting for Server { let object_id = ObjectType::DmarcInternalReport.to_id(); let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id }); - let Some(report) = self - .store() - .get_value::(ValueKey::from(key.clone())) - .await - .caused_by(trc::location!())? - else { - return Ok(()); - }; + // Delete report. inbuxa: only the version read here, so a record + // another node appends meanwhile is sent with it rather than lost + let mut attempt = 0; + let report = loop { + let Some(Revisioned { + revision, + value: report, + }) = self + .store() + .get_value::>(ValueKey::from(key.clone())) + .await + .caused_by(trc::location!())? + else { + return Ok(()); + }; - // Delete report - let mut batch = BatchBuilder::new(); - batch.clear(key).clear(RegistryClass::PrimaryKey { - object_id: object_id.into(), - index_id: Property::Domain.to_id(), - key: KeySerializer::new(report.domain.len() + U64_LEN) - .write(&report.domain) - .write(report.policy_identifier) - .finalize(), - }); - self.store() - .write(batch.build_all()) - .await - .caused_by(trc::location!())?; + let mut batch = BatchBuilder::new(); + batch + .assert_value(key.clone(), AssertValue::Hash(revision)) + .clear(key.clone()) + .clear(RegistryClass::PrimaryKey { + object_id: object_id.into(), + index_id: Property::Domain.to_id(), + key: KeySerializer::new(report.domain.len() + U64_LEN) + .write(&report.domain) + .write(report.policy_identifier) + .finalize(), + }); + match self.store().write(batch.build_all()).await { + Ok(_) => break report, + Err(err) if err.is_assertion_failure() && attempt < MAX_WRITE_RETRIES => { + attempt += 1; + write_retry_pause(attempt).await; + } + Err(err) => return Err(err.caused_by(trc::location!())), + } + }; let span_id = self.inner.data.span_id_gen.generate(); let event_from = report.report.date_range_begin.timestamp() as u64; @@ -676,8 +693,11 @@ impl DmarcReporting for Server { break; } Err(err) => { - if err.is_assertion_failure() && rety_count < 3 { + // inbuxa: another node appended first; try again + // after a short pause + if err.is_assertion_failure() && rety_count < MAX_WRITE_RETRIES { rety_count += 1; + write_retry_pause(rety_count).await; continue; } trc::error!( diff --git a/crates/smtp/src/reporting/mod.rs b/crates/smtp/src/reporting/mod.rs index add57eb..20e2064 100644 --- a/crates/smtp/src/reporting/mod.rs +++ b/crates/smtp/src/reporting/mod.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 common::config::smtp::report::AggregateFrequency; @@ -15,6 +17,7 @@ pub mod inbound; pub mod index; pub mod scheduler; pub mod send; +pub mod shared; // inbuxa: reports written by every node pub mod spf; pub mod tls; diff --git a/crates/smtp/src/reporting/scheduler.rs b/crates/smtp/src/reporting/scheduler.rs index 4b08f7b..ed4fec4 100644 --- a/crates/smtp/src/reporting/scheduler.rs +++ b/crates/smtp/src/reporting/scheduler.rs @@ -20,14 +20,17 @@ 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; - } + // inbuxa: every node records what it received, whatever its + // role. An aggregate report covers all of a domain's mail, + // whichever node took it, and recording is a store write + // that nodes already share: the report's primary key is + // versioned, so concurrent appends from several nodes retry + // rather than overwrite. Only building and sending the + // report (the DmarcReport and TlsReport tasks) belongs to + // the outbound MTA; the task manager keeps those to nodes + // with that role. Upstream ran this only on outbound MTA + // nodes, so mail received anywhere else never reached a + // report. match event { ReportingEvent::Dmarc(event) => server.schedule_dmarc(event).await, ReportingEvent::Tls(event) => server.schedule_tls(event).await, diff --git a/crates/smtp/src/reporting/shared.rs b/crates/smtp/src/reporting/shared.rs new file mode 100644 index 0000000..258ceda --- /dev/null +++ b/crates/smtp/src/reporting/shared.rs @@ -0,0 +1,45 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! inbuxa: internal DMARC and TLS reports are shared by every node. Any node +//! that receives mail appends to them, so several nodes can write one report +//! at once, and the node that sends it may do so while another is appending. +//! Appends already guard the report's versioned primary key and retry when +//! another writer got there first; these helpers give those retries room and +//! let the sender delete exactly the report it read. + +use rand::RngExt; +use std::time::Duration; +use store::{Deserialize, xxhash_rust::xxh3::xxh3_64}; + +/// How many times a report write that lost to another writer is retried. +/// Upstream retried three times, when only outbound MTA nodes wrote. +pub(crate) const MAX_WRITE_RETRIES: u32 = 10; + +/// A short random pause, longer on each attempt, before retrying a report +/// write that lost to another node, so the writers spread out instead of +/// colliding again. +pub(crate) async fn write_retry_pause(attempt: u32) { + let ms = rand::rng().random_range(5..=25u64) * u64::from(attempt.max(1)); + tokio::time::sleep(Duration::from_millis(ms)).await; +} + +/// A stored value with the hash of the bytes it was read from, for +/// `AssertValue::Hash`: a write asserting it fails if anyone changed the +/// value since. +pub(crate) struct Revisioned { + pub revision: u64, + pub value: T, +} + +impl Deserialize for Revisioned { + fn deserialize(bytes: &[u8]) -> trc::Result { + Ok(Revisioned { + revision: xxh3_64(bytes), + value: T::deserialize(bytes)?, + }) + } +} diff --git a/crates/smtp/src/reporting/tls.rs b/crates/smtp/src/reporting/tls.rs index 3842407..f489e43 100644 --- a/crates/smtp/src/reporting/tls.rs +++ b/crates/smtp/src/reporting/tls.rs @@ -2,9 +2,12 @@ * 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::AggregateTimestamp; +use super::shared::{MAX_WRITE_RETRIES, Revisioned, write_retry_pause}; use crate::{ queue::RecipientDomain, reporting::{index::InternalReportIndex, send::MtaReportSend}, @@ -70,28 +73,40 @@ impl TlsReporting for Server { let object_id = ObjectType::TlsInternalReport.to_id(); let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id }); - let Some(report) = self - .store() - .get_value::(ValueKey::from(key.clone())) - .await - .caused_by(trc::location!())? - else { - return Ok(()); - }; + // Delete report. inbuxa: only the version read here, so a result + // another node appends meanwhile is sent with it rather than lost + let mut attempt = 0; + let report = loop { + let Some(Revisioned { + revision, + value: report, + }) = self + .store() + .get_value::>(ValueKey::from(key.clone())) + .await + .caused_by(trc::location!())? + else { + return Ok(()); + }; - // Delete report - let mut batch = BatchBuilder::new(); - batch.clear(key).clear(RegistryClass::PrimaryKey { - object_id: object_id.into(), - index_id: Property::Domain.to_id(), - key: report.domain.as_bytes().to_vec(), - }); - self.core - .storage - .data - .write(batch.build_all()) - .await - .caused_by(trc::location!())?; + let mut batch = BatchBuilder::new(); + batch + .assert_value(key.clone(), AssertValue::Hash(revision)) + .clear(key.clone()) + .clear(RegistryClass::PrimaryKey { + object_id: object_id.into(), + index_id: Property::Domain.to_id(), + key: report.domain.as_bytes().to_vec(), + }); + match self.core.storage.data.write(batch.build_all()).await { + Ok(_) => break report, + Err(err) if err.is_assertion_failure() && attempt < MAX_WRITE_RETRIES => { + attempt += 1; + write_retry_pause(attempt).await; + } + Err(err) => return Err(err.caused_by(trc::location!())), + } + }; let domain_name = report.domain.as_str(); let event_from = report.report.date_range_start.timestamp() as u64; @@ -477,8 +492,11 @@ impl TlsReporting for Server { break; } Err(err) => { - if err.is_assertion_failure() && rety_count < 3 { + // inbuxa: another node appended first; try again + // after a short pause + if err.is_assertion_failure() && rety_count < MAX_WRITE_RETRIES { rety_count += 1; + write_retry_pause(rety_count).await; continue; } trc::error!( diff --git a/tests/src/cluster/front_reports.rs b/tests/src/cluster/front_reports.rs new file mode 100644 index 0000000..f702042 --- /dev/null +++ b/tests/src/cluster/front_reports.rs @@ -0,0 +1,317 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! DMARC results recorded on a node without outboundMta reach the aggregate +//! report, which a node with outboundMta builds and sends. Before, a front +//! node's results were dropped (or, before live roles, left in a channel +//! nobody read), so the report covered only the mail the outbound nodes +//! received. Also checks that nodes appending to one report at once lose +//! nothing. Needs a store the nodes can share (STORE=PostgreSql or MySql). + +use crate::{smtp::inbound::TestMessage, utils::server::TestServerBuilder}; +use common::{Server, config::smtp::report::AggregateFrequency, ipc::DmarcEvent}; +use mail_auth::{ + common::parse::TxtRecordParser, + dmarc::Dmarc, + report::{ActionDisposition, DmarcResult, Record, Report}, +}; +use registry::{ + schema::{ + enums::ClusterTaskType, + prelude::{ObjectType, Property}, + structs::{ + ClusterListenerGroup, ClusterRole, ClusterTaskGroup, ClusterTaskGroupProperties, + DmarcInternalReport, DmarcReportSettings, Expression, Task, TaskDmarcReport, + TaskStatus, + }, + }, + types::{EnumImpl, map::Map}, +}; +use smtp::reporting::{dmarc::DmarcReporting, send::MtaReportSend}; +use std::{ + collections::BTreeSet, + net::IpAddr, + sync::Arc, + time::{Duration, Instant}, +}; +use store::{ + ValueKey, + registry::{RegistryFilter, RegistryFilterValue, RegistryQuery}, + write::{BatchBuilder, RegistryClass, TaskQueueClass, ValueClass, now}, +}; +use types::id::Id; + +const FRONT_ROLE: &str = "front_reports_front"; +const MTA_ROLE: &str = "front_reports_mta"; +const DOMAIN: &str = "front-reports.example"; + +#[tokio::test(flavor = "multi_thread")] +pub async fn front_node_report_tests() { + if matches!( + std::env::var("STORE").as_deref(), + Ok("RocksDb" | "Sqlite") | Err(_) + ) { + println!("Skipping front node report tests: they need a store the nodes can share."); + return; + } + println!( + "Running front node report tests on {}...", + std::env::var("STORE").unwrap_or_default() + ); + + // A front role without outboundMta, an MTA role with it + let seed = TestServerBuilder::new("front_reports_seed").await; + seed.insert_object(role(FRONT_ROLE, &[ClusterTaskType::PushNotifications])) + .await; + seed.insert_object(role(MTA_ROLE, &[ClusterTaskType::OutboundMta])) + .await; + seed.insert_object(DmarcReportSettings { + aggregate_max_report_size: Expression { + else_: "1048576".into(), + ..Default::default() + }, + ..Default::default() + }) + .await; + let seed = seed.disable_services().build().await; + + // The front node receives mail from two sources: the events its SMTP + // sessions hand the report scheduler + let front = TestServerBuilder::new_with_role( + "front_reports_front", + "front.front-reports.example".into(), + Some(FRONT_ROLE.into()), + false, + ) + .await + .build_with_opts(false) + .await; + let front_server = front.server.clone(); + assert!(!front_server.core.network.roles.outbound_mta); + for ip in ["192.0.2.1", "192.0.2.2"] { + front_server.schedule_report(event(ip)).await; + } + + // Both are recorded in the shared report. Upstream, and main after live + // roles, left the front node's results out + let report_id = wait_for_report(&front_server, 2).await; + + // Make the report due now. The front node leaves it alone: building and + // sending it is the outbound MTA's + move_task(&front_server, report_id, TaskStatus::now()).await; + tokio::time::sleep(Duration::from_secs(3)).await; + front_server.notify_task_queue(); + tokio::time::sleep(Duration::from_secs(2)).await; + assert!( + task_exists(&front_server, report_id).await, + "the front node ran the report task" + ); + + // Several writers append to the report at once, from both nodes: none + // of their records is lost. The report waits in the future meanwhile, or + // the MTA node would send it as soon as it starts + move_task( + &front_server, + report_id, + TaskStatus::at(now() as i64 + 3600), + ) + .await; + let mut mta = TestServerBuilder::new_with_role( + "front_reports_mta", + "mta.front-reports.example".into(), + Some(MTA_ROLE.into()), + false, + ) + .await + .capture_queue() + .build_with_opts(false) + .await; + let mta_server = mta.server.clone(); + assert!(mta_server.core.network.roles.outbound_mta); + + let concurrent: Vec = (10..18).map(|n| format!("192.0.2.{n}")).collect(); + let mut handles = Vec::new(); + for (n, ip) in concurrent.iter().enumerate() { + let server = if n % 2 == 0 { + front_server.clone() + } else { + mta_server.clone() + }; + let ip = ip.clone(); + handles.push(tokio::spawn(async move { + server.schedule_dmarc(Box::new(event(&ip))).await; + })); + } + for handle in handles { + handle.await.unwrap(); + } + + // Due again, the MTA node sends the report with every record in it + move_task(&mta_server, report_id, TaskStatus::now()).await; + let message = mta.expect_message().await; + let report = + Report::parse_rfc5322(message.read_message(&mta).await.as_bytes(), usize::MAX).unwrap(); + assert_eq!(report.domain(), DOMAIN); + let sent: BTreeSet = report + .records() + .iter() + .map(|r| r.source_ip().unwrap()) + .collect(); + let expected: BTreeSet = ["192.0.2.1", "192.0.2.2"] + .into_iter() + .map(String::from) + .chain(concurrent) + .map(|ip| ip.parse().unwrap()) + .collect(); + assert_eq!(sent, expected); + wait_for(Duration::from_secs(20), "report task to finish", || async { + !task_exists(&mta_server, report_id).await + }) + .await; + assert!(reports(&mta_server).await.is_empty()); + + if seed.is_reset() { + seed.temp_dir.delete(); + front.temp_dir.delete(); + mta.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 event(ip: &str) -> DmarcEvent { + DmarcEvent { + domain: DOMAIN.to_string(), + report_record: Record::new() + .with_source_ip(ip.parse().unwrap()) + .with_action_disposition(ActionDisposition::Pass) + .with_dmarc_dkim_result(DmarcResult::Pass) + .with_dmarc_spf_result(DmarcResult::Pass) + .with_envelope_from("sender.example") + .with_header_from("sender.example"), + dmarc_record: Arc::new( + Dmarc::parse(format!("v=DMARC1; p=reject; rua=mailto:reports@{DOMAIN}").as_bytes()) + .unwrap(), + ), + interval: AggregateFrequency::Daily, + span_id: 0, + } +} + +async fn reports(server: &Server) -> Vec<(u64, DmarcInternalReport)> { + let ids = server + .registry() + .query::>(RegistryQuery::new(ObjectType::DmarcInternalReport).filter( + RegistryFilter::greater_than( + Property::Domain, + RegistryFilterValue::Bytes(vec![]), + true, + ), + )) + .await + .unwrap(); + let mut reports = Vec::new(); + for id in ids { + if let Some(report) = server + .store() + .get_value::(ValueKey::from(ValueClass::Registry( + RegistryClass::Item { + object_id: ObjectType::DmarcInternalReport.to_id(), + item_id: id.id(), + }, + ))) + .await + .unwrap() + { + reports.push((id.id(), report)); + } + } + reports +} + +/// Waits for the report for `DOMAIN` to hold `records` records; returns its id. +async fn wait_for_report(server: &Server, records: usize) -> u64 { + let started = Instant::now(); + loop { + let found = reports(server) + .await + .into_iter() + .find(|(_, report)| report.domain == DOMAIN); + if let Some((id, report)) = &found + && report.report.records.len() == records + { + return *id; + } + assert!( + started.elapsed() < Duration::from_secs(10), + "no report with {records} records for {DOMAIN}: {found:?}" + ); + tokio::time::sleep(Duration::from_millis(200)).await; + } +} + +/// Reschedules the report's task. +async fn move_task(server: &Server, id: u64, status: TaskStatus) { + let task = server + .store() + .get_value::(ValueKey::from(ValueClass::TaskQueue( + TaskQueueClass::Task { id }, + ))) + .await + .unwrap() + .expect("report task missing"); + let mut batch = BatchBuilder::new(); + batch + .clear(ValueClass::TaskQueue(TaskQueueClass::Due { + id, + due: task.due_timestamp(), + })) + .schedule_task_with_id( + id, + Task::DmarcReport(TaskDmarcReport { + report_id: id.into(), + status, + }), + ); + server.store().write(batch.build_all()).await.unwrap(); + server.notify_task_queue(); +} + +async fn task_exists(server: &Server, id: u64) -> bool { + server + .store() + .get_value::(ValueKey::from(ValueClass::TaskQueue( + TaskQueueClass::Task { id }, + ))) + .await + .unwrap() + .is_some() +} + +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 ae79136..a9be0fb 100644 --- a/tests/src/cluster/mod.rs +++ b/tests/src/cluster/mod.rs @@ -7,6 +7,7 @@ */ pub mod broadcast; +pub mod front_reports; // inbuxa: every node records DMARC and TLS results pub mod live_roles; // inbuxa: role edits apply without a restart #[cfg(feature = "nats")] pub mod coordinator; // inbuxa: coordinator reconnects