Fix MTA: A node without the outboundMta role stops replying to DATA and to JMAP submissions once about 1024 messages have been queued on it

CHANGELOG.md

@@ -24,6 +24,7 @@ If you are upgrading from v0.16.x, replace the binary (or run `docker pull`). If
   - Changes made while a push request is in flight are not delivered until the next change reaches the same subscription, since a successful delivery cancels the pending retry.
   - The VAPID `aud` claim is derived from a hand-written parse of the push URL, so a crafted push URL can make the server sign a token for a push service other than the one the request is sent to.
   - `Email/import` rejects a `blobId` that refers to a `Blob/upload` creation id in the same request (`"#u0"`) with `Invalid blob id.`.
+- MTA: A node without the `outboundMta` role stops replying to `DATA` and to JMAP submissions once about 1024 messages have been queued on it.
 
 ## [0.16.23] - 2026-09-21
 

crates/smtp/src/lib.rs

@@ -38,12 +38,19 @@ impl StartQueueManager for BootManager {
 impl SpawnQueueManager for IpcReceivers {
     fn spawn_queue_manager(&mut self, inner: Arc<Inner>) {
         let core = inner.shared_core.load();
-        if !core.storage.registry.is_recovery_mode() && core.network.roles.outbound_mta {
-            // Spawn queue manager
-            self.queue_rx.take().unwrap().spawn(inner.clone());
-
-            // Spawn report manager
-            self.report_rx.take().unwrap().spawn(inner);
+        if core.storage.registry.is_recovery_mode() {
+            return;
         }
+
+        // Spawn queue manager
+        let queue_rx = self.queue_rx.take().unwrap();
+        if core.network.roles.outbound_mta {
+            queue_rx.spawn(inner.clone());
+        } else {
+            queue_rx.discard();
+        }
+
+        // Spawn report manager
+        self.report_rx.take().unwrap().spawn(inner);
     }
 }

crates/smtp/src/queue/manager.rs

@@ -62,6 +62,10 @@ impl SpawnQueue for mpsc::Receiver<QueueEvent> {
             Queue::new(core, self).start().await;
         });
     }
+
+    fn discard(mut self) {
+        tokio::spawn(async move { while self.recv().await.is_some() {} });
+    }
 }
 
 const BACK_PRESSURE_WARN_INTERVAL: Duration = Duration::from_secs(60);
@@ -480,6 +484,7 @@ impl Recipient {
 
 pub trait SpawnQueue {
     fn spawn(self, core: Arc<Inner>);
+    fn discard(self);
 }
 
 impl QueueStats {