Every node records DMARC and TLS results for the aggregate reports

The report scheduler dropped DMARC and TLS events on a node whose role
lacks outboundMta (upstream never started it there, so they sat in a
channel nobody read). Mail received on a front node therefore never
reached an aggregate report, which is meant to cover all of a domain's
inbound mail, whichever node received it. In rehearsal, five messages
received on port 25 on a front node were missing from every report.

- The report scheduler records on every node. Recording is a store write
  the nodes already share, so it needs nothing from the outbound MTA.
  Building and sending a report (the DmarcReport and TlsReport tasks) stay
  with outboundMta nodes, as the task manager already enforces.
- More nodes now append to one report at once. Appends already guard the
  report's versioned primary key; a write that loses now retries up to ten
  times after a short random pause, not three times at once.
- The node sending a report deletes it only if it is unchanged since it
  was read, and reads it again otherwise, so a record another node appends
  meanwhile goes out with the report instead of being deleted unsent.

Test: cluster::front_reports (PostgreSQL and MySQL). A front node's
results appear in the report the MTA node sends, alongside eight appended
at once from both nodes, and the front node never runs the report task.
It fails on main: the front node's results are never recorded.

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
+        // on every node: the queue follows the role live (see Queue::start),
-        // report scheduler). This also drains the queue channel on nodes
+        // 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() {

crates/smtp/src/reporting/dmarc.rs

@@ -2,9 +2,12 @@
  * 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::AggregateTimestamp;
+use super::shared::{MAX_WRITE_RETRIES, Revisioned, write_retry_pause};
 use crate::{
     core::Session,
     queue::RecipientDomain,
@@ -349,18 +352,27 @@ 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
+        // 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::<DmarcInternalReport>(ValueKey::from(key.clone()))
+                .get_value::<Revisioned<DmarcInternalReport>>(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 {
+            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)
@@ -368,10 +380,15 @@ impl DmarcReporting for Server {
                         .write(report.policy_identifier)
                         .finalize(),
                 });
-        self.store()
+            match self.store().write(batch.build_all()).await {
-            .write(batch.build_all())
+                Ok(_) => break report,
-            .await
+                Err(err) if err.is_assertion_failure() && attempt < MAX_WRITE_RETRIES => {
-            .caused_by(trc::location!())?;
+                    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!(

crates/smtp/src/reporting/mod.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 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;
 

crates/smtp/src/reporting/scheduler.rs

@@ -20,14 +20,17 @@ 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
+                // inbuxa: every node records what it received, whatever its
-                // boot, but the role is read per event so a change applies
+                // role. An aggregate report covers all of a domain's mail,
-                // without a restart. Events that arrive while the role is
+                // whichever node took it, and recording is a store write
-                // off are dropped, as they were on a node started without it
+                // that nodes already share: the report's primary key is
-                if !matches!(event, ReportingEvent::Stop) && !server.core.network.roles.outbound_mta
+                // versioned, so concurrent appends from several nodes retry
-                {
+                // rather than overwrite. Only building and sending the
-                    continue;
+                // 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,

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<T> {
+    pub revision: u64,
+    pub value: T,
+}
+
+impl<T: Deserialize> Deserialize for Revisioned<T> {
+    fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
+        Ok(Revisioned {
+            revision: xxh3_64(bytes),
+            value: T::deserialize(bytes)?,
+        })
+    }
+}

crates/smtp/src/reporting/tls.rs

@@ -2,9 +2,12 @@
  * 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::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
+        // 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::<TlsInternalReport>(ValueKey::from(key.clone()))
+                .get_value::<Revisioned<TlsInternalReport>>(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 {
+            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(),
                 });
-        self.core
+            match self.core.storage.data.write(batch.build_all()).await {
-            .storage
+                Ok(_) => break report,
-            .data
+                Err(err) if err.is_assertion_failure() && attempt < MAX_WRITE_RETRIES => {
-            .write(batch.build_all())
+                    attempt += 1;
-            .await
+                    write_retry_pause(attempt).await;
-            .caused_by(trc::location!())?;
+                }
+                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!(

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<String> = (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<IpAddr> = report
+        .records()
+        .iter()
+        .map(|r| r.source_ip().unwrap())
+        .collect();
+    let expected: BTreeSet<IpAddr> = ["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::<Vec<Id>>(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::<DmarcInternalReport>(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::<Task>(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::<Task>(ValueKey::from(ValueClass::TaskQueue(
+            TaskQueueClass::Task { id },
+        )))
+        .await
+        .unwrap()
+        .is_some()
+}
+
+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 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