Bootstrapping implementation

crates/smtp/src/inbound/auth.rs

@@ -59,11 +59,9 @@ impl<T: SessionStream> Session<T> {
                     Credentials::Basic {
                         username, secret, ..
                     },
-                ) => {
-                    if username.is_empty() && secret.is_empty() {
-                        self.write(b"334 VXNlcm5hbWU6\r\n").await?;
-                        return Ok(true);
-                    }
+                ) if username.is_empty() && secret.is_empty() => {
+                    self.write(b"334 VXNlcm5hbWU6\r\n").await?;
+                    return Ok(true);
                 }
                 _ => (),
             }
@@ -71,7 +69,7 @@ impl<T: SessionStream> Session<T> {
             match (token.mechanism, &mut token.credentials) {
                 (AUTH_PLAIN, _) => {
                     if let Some(credentials) = Credentials::decode_sasl_challenge_plain(&response) {
-                        return self.authenticate(credentials).await;
+                        return Box::pin(self.authenticate(credentials)).await;
                     }
                 }
                 (
@@ -86,20 +84,20 @@ impl<T: SessionStream> Session<T> {
                         Ok(true)
                     } else {
                         *secret = response.into_string();
-                        self.authenticate(std::mem::replace(
+                        Box::pin(self.authenticate(std::mem::replace(
                             &mut token.credentials,
                             Credentials::Basic {
                                 username: String::new(),
                                 secret: String::new(),
                                 mfa_token: None,
                             },
-                        ))
+                        )))
                         .await
                     };
                 }
                 (AUTH_OAUTHBEARER | AUTH_XOAUTH2, _) => {
                     if let Some(credentials) = Credentials::decode_sasl_challenge_oauth(&response) {
-                        return self.authenticate(credentials).await;
+                        return Box::pin(self.authenticate(credentials)).await;
                     }
                 }
                 _ => (),

crates/smtp/src/inbound/mail.rs

@@ -171,12 +171,10 @@ impl<T: SessionStream> Session<T> {
                 )
                 .await
             {
-                ScriptResult::Accept { modifications } => {
-                    if !modifications.is_empty() {
-                        for modification in modifications {
-                            if let ScriptModification::SetEnvelope { name, value } = modification {
-                                self.data.apply_envelope_modification(name, value);
-                            }
+                ScriptResult::Accept { modifications } if !modifications.is_empty() => {
+                    for modification in modifications {
+                        if let ScriptModification::SetEnvelope { name, value } = modification {
+                            self.data.apply_envelope_modification(name, value);
                         }
                     }
                 }

crates/smtp/src/inbound/rcpt.rs

@@ -124,14 +124,10 @@ impl<T: SessionStream> Session<T> {
                     )
                     .await
                 {
-                    ScriptResult::Accept { modifications } => {
-                        if !modifications.is_empty() {
-                            for modification in modifications {
-                                if let ScriptModification::SetEnvelope { name, value } =
-                                    modification
-                                {
-                                    self.data.apply_envelope_modification(name, value);
-                                }
+                    ScriptResult::Accept { modifications } if !modifications.is_empty() => {
+                        for modification in modifications {
+                            if let ScriptModification::SetEnvelope { name, value } = modification {
+                                self.data.apply_envelope_modification(name, value);
                             }
                         }
                     }

crates/smtp/src/inbound/spawn.rs

@@ -153,7 +153,7 @@ impl<T: SessionStream> Session<T> {
                                 if bytes_read > 0 {
                                     if Instant::now() < self.data.valid_until && bytes_read <= self.data.bytes_left  {
                                         self.data.bytes_left -= bytes_read;
-                                        match self.ingest(&buf[..bytes_read]).await {
+                                        match Box::pin(self.ingest(&buf[..bytes_read])).await {
                                             Ok(true) => (),
                                             Ok(false) => {
                                                 return true;

crates/smtp/src/lib.rs

@@ -7,7 +7,7 @@
 #![warn(clippy::large_futures)]
 
 use common::{
-    BuildServer, Inner,
+    Inner,
     manager::boot::{BootManager, IpcReceivers},
 };
 use queue::manager::SpawnQueue;
@@ -37,7 +37,8 @@ impl StartQueueManager for BootManager {
 
 impl SpawnQueueManager for IpcReceivers {
     fn spawn_queue_manager(&mut self, inner: Arc<Inner>) {
-        if inner.build_server().core.network.roles.outbound_mta {
+        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());
 

crates/smtp/src/queue/manager.rs

@@ -168,6 +168,7 @@ impl Queue {
 
                 match status {
                     QueueEventStatus::Completed => {
+                        self.core.ipc.task_tx.notify_one();
                         self.locked.remove(&(queue_id, queue_name));
                         !self.locked.is_empty() || !queue_stats.has_capacity()
                     }