Fix MTA: Always update next DSN notify times

crates/smtp/src/queue/dsn.rs

@@ -62,6 +62,9 @@ impl SendDsn for Server {
             // Handle double bounce
             message.handle_double_bounce();
         }
+
+        // Update next DSN notify times
+        message.update_next_dsn(self).await;
     }
 
     async fn log_dsn(&self, message: &MessageWrapper) {
@@ -186,7 +189,6 @@ impl MessageWrapper {
             dsn.push_str("\r\n");
         }
 
-        // Build text response
         let txt_len = txt_success.len() + txt_delay.len() + txt_failed.len();
         if txt_len == 0 {
             return None;
@@ -250,44 +252,6 @@ impl MessageWrapper {
             txt.push_str("\r\n");
         }
 
-        // Update next delay notification time
-        if has_delay {
-            let mut changes = Vec::new();
-            for (rcpt_idx, rcpt) in self.message.recipients.iter().enumerate() {
-                if matches!(
-                    &rcpt.status,
-                    Status::TemporaryFailure(_) | Status::Scheduled
-                ) && rcpt.notify.due <= now
-                {
-                    let envelope = QueueEnvelope::new(&self.message, rcpt);
-
-                    let queue_id = server
-                        .eval_if::<String, _>(
-                            &server.core.smtp.queue.queue,
-                            &envelope,
-                            self.span_id,
-                        )
-                        .await
-                        .unwrap_or_else(|| "default".to_string());
-                    let queue = server.get_queue_or_default(&queue_id, self.span_id);
-
-                    if let Some(next_notify) =
-                        queue.notify.get((rcpt.notify.inner + 1) as usize).copied()
-                    {
-                        changes.push((rcpt_idx, 1, now + next_notify));
-                    } else {
-                        changes.push((rcpt_idx, 0, u64::MAX));
-                    }
-                }
-            }
-
-            for (rcpt_idx, inner, due) in changes {
-                let rcpt = &mut self.message.recipients[rcpt_idx];
-                rcpt.notify.inner += inner;
-                rcpt.notify.due = due;
-            }
-        }
-
         // Obtain hostname and sender addresses
         let from_name = server
             .eval_if(&config.dsn.name, &self.message, self.span_id)
@@ -393,6 +357,40 @@ impl MessageWrapper {
             .into()
     }
 
+    pub async fn update_next_dsn(&mut self, server: &Server) {
+        let now = now();
+        let mut notify_changes = Vec::new();
+        for (rcpt_idx, rcpt) in self.message.recipients.iter().enumerate() {
+            if matches!(
+                &rcpt.status,
+                Status::TemporaryFailure(_) | Status::Scheduled
+            ) && rcpt.notify.due <= now
+            {
+                let envelope = QueueEnvelope::new(&self.message, rcpt);
+
+                let queue_id = server
+                    .eval_if::<String, _>(&server.core.smtp.queue.queue, &envelope, self.span_id)
+                    .await
+                    .unwrap_or_else(|| "default".to_string());
+                let queue = server.get_queue_or_default(&queue_id, self.span_id);
+
+                if let Some(next_notify) =
+                    queue.notify.get((rcpt.notify.inner + 1) as usize).copied()
+                {
+                    notify_changes.push((rcpt_idx, 1, now + next_notify));
+                } else {
+                    notify_changes.push((rcpt_idx, 0, u64::MAX));
+                }
+            }
+        }
+
+        for (rcpt_idx, inner, due) in notify_changes {
+            let rcpt = &mut self.message.recipients[rcpt_idx];
+            rcpt.notify.inner += inner;
+            rcpt.notify.due = due;
+        }
+    }
+
     fn handle_double_bounce(&mut self) {
         let mut is_double_bounce = Vec::with_capacity(0);
         let now = now();

crates/smtp/src/queue/spool.rs

@@ -14,7 +14,7 @@ use crate::queue::{
     FROM_UNAUTHENTICATED_DMARC, MessageWrapper,
 };
 use ahash::AHashSet;
-use common::config::smtp::queue::QueueName;
+use common::config::smtp::queue::{ArchivedQueueExpiry, QueueName};
 use common::ipc::QueueEvent;
 use common::{KV_LOCK_QUEUE_MESSAGE, Server};
 use registry::schema::prelude::{ObjectType, Property};
@@ -803,6 +803,35 @@ impl ArchivedMessage {
         next_delivery
     }
 
+    pub fn next_event(&self, queue: Option<QueueName>) -> Option<u64> {
+        let created = self.created.to_native();
+        let mut next_event = None;
+
+        for rcpt in self.recipients.iter().filter(|d| {
+            matches!(
+                d.status,
+                ArchivedStatus::Scheduled | ArchivedStatus::TemporaryFailure(_)
+            ) && queue.is_none_or(|q| d.queue == q)
+        }) {
+            let mut earlier_event =
+                std::cmp::min(rcpt.retry.due.to_native(), rcpt.notify.due.to_native());
+
+            if let ArchivedQueueExpiry::Ttl(ttl) = &rcpt.expires {
+                earlier_event = std::cmp::min(earlier_event, created + ttl.to_native());
+            }
+
+            if let Some(next_event) = &mut next_event {
+                if earlier_event < *next_event {
+                    *next_event = earlier_event;
+                }
+            } else {
+                next_event = Some(earlier_event);
+            }
+        }
+
+        next_event
+    }
+
     pub fn next_notify_event(&self, queue: Option<QueueName>) -> Option<u64> {
         let mut next_notify = None;