fix scheduler deadlock when core count is low

The GH actions CI hangs were due to a deadlock in rt_spawn.
The GH runners have only two cores available which meant
that it would only take one call to send_blocking or recv_blocking
to tie up the localset threads and deadlock.

This commit adjusts that logic to make rt_spawn async and to
provide a non-blocking fire and forget variant for the couple
of situations that need it.

Also adds a bit of tracing!

crates/integration-tests/src/kumod.rs

@@ -51,9 +51,8 @@ impl KumoDaemon {
 
         let mut child = Command::new(&path)
             .args(["--policy", &args.policy_file])
-            .env("KUMOD_LOG", "trace")
+            .env("KUMOD_LOG", "kumod=trace")
             .env("KUMOD_TEST_DIR", dir.path())
-            .env("TOKIO_CONSOLE_BIND", "127.0.0.1:0")
             .envs(args.env.iter().cloned())
             .stdout(Stdio::piped())
             .stderr(Stdio::piped())
@@ -129,6 +128,7 @@ impl KumoDaemon {
 
         let banner = client.read_response(None).await?;
         anyhow::ensure!(banner.code == 220, "unexpected banner: {banner:#?}");
+        client.ehlo("localhost").await?;
         Ok(client)
     }
 

crates/integration-tests/src/main.rs

@@ -27,7 +27,6 @@ mod test {
         eprintln!("sending message");
         tokio::time::timeout(Duration::from_secs(10), async {
             let mut client = source.smtp_client().await?;
-            client.ehlo("localhost").await?;
             const BODY: &str = "From: <me@localhost>\r\n\
                                 To: <you@localhost>\r\n\
                                 Subject: a test message\r\n\

crates/kumod/src/egress_path.rs

@@ -5,7 +5,7 @@ use crate::lifecycle::{Activity, ShutdownSubcription};
 use crate::logging::{log_disposition, LogDisposition, RecordType};
 use crate::mx::{MailExchanger, ResolvedAddress};
 use crate::queue::{Queue, QueueManager};
-use crate::runtime::{rt_spawn, spawn};
+use crate::runtime::{rt_spawn, rt_spawn_non_blocking, spawn};
 use crate::spool::SpoolManager;
 use anyhow::Context;
 use config::load_config;
@@ -206,7 +206,7 @@ impl EgressPathManager {
         let mut manager = Self::get().await;
         let activity = Activity::get()?;
         let handle = manager.paths.entry(name.clone()).or_insert_with(|| {
-            rt_spawn(format!("maintain {name}"), {
+            rt_spawn_non_blocking(format!("maintain {name}"), {
                 let name = name.clone();
                 move || {
                     Ok(async move {
@@ -216,7 +216,7 @@ impl EgressPathManager {
                             tokio::select! {
                                 _ = tokio::time::sleep(interval) => {},
                                 _ = shutdown.shutting_down() => {
-                                    interval = Duration::from_secs(5);
+                                    interval = Duration::from_secs(1);
                                 }
                             };
                             let mut mgr = EgressPathManager::get().await;
@@ -494,34 +494,38 @@ impl Dispatcher {
         };
 
         if dispatcher.addresses.is_empty() {
-            dispatcher.bulk_ready_queue_operation(Response {
-                code: 556,
-                enhanced_code: Some(EnhancedStatusCode {
-                    class: 5,
-                    subject: 1,
-                    detail: 10,
-                }),
-                content: "Recipient address has a null MX".to_string(),
-                command: None,
-            });
+            dispatcher
+                .bulk_ready_queue_operation(Response {
+                    code: 556,
+                    enhanced_code: Some(EnhancedStatusCode {
+                        class: 5,
+                        subject: 1,
+                        detail: 10,
+                    }),
+                    content: "Recipient address has a null MX".to_string(),
+                    command: None,
+                })
+                .await;
             return Ok(());
         }
 
         for addr in &dispatcher.addresses {
             if dispatcher.path_config.prohibited_hosts.contains(addr.addr) {
-                dispatcher.bulk_ready_queue_operation(Response {
-                    code: 550,
-                    enhanced_code: Some(EnhancedStatusCode {
-                        class: 5,
-                        subject: 4,
-                        detail: 4,
-                    }),
-                    content: format!(
-                        "{addr:?} is on the list of prohibited_hosts {:?}",
-                        dispatcher.path_config.prohibited_hosts
-                    ),
-                    command: None,
-                });
+                dispatcher
+                    .bulk_ready_queue_operation(Response {
+                        code: 550,
+                        enhanced_code: Some(EnhancedStatusCode {
+                            class: 5,
+                            subject: 4,
+                            detail: 4,
+                        }),
+                        content: format!(
+                            "{addr:?} is on the list of prohibited_hosts {:?}",
+                            dispatcher.path_config.prohibited_hosts
+                        ),
+                        command: None,
+                    })
+                    .await;
                 return Ok(());
             }
         }
@@ -531,16 +535,18 @@ impl Dispatcher {
             .retain(|addr| !dispatcher.path_config.skip_hosts.contains(addr.addr));
 
         if dispatcher.addresses.is_empty() {
-            dispatcher.bulk_ready_queue_operation(Response {
-                code: 550,
-                enhanced_code: Some(EnhancedStatusCode {
-                    class: 5,
-                    subject: 4,
-                    detail: 4,
-                }),
-                content: "MX consisted solely of hosts on the skip_hosts list".to_string(),
-                command: None,
-            });
+            dispatcher
+                .bulk_ready_queue_operation(Response {
+                    code: 550,
+                    enhanced_code: Some(EnhancedStatusCode {
+                        class: 5,
+                        subject: 4,
+                        detail: 4,
+                    }),
+                    content: "MX consisted solely of hosts on the skip_hosts list".to_string(),
+                    command: None,
+                })
+                .await;
             return Ok(());
         }
 
@@ -567,7 +573,7 @@ impl Dispatcher {
                             .path_config
                             .consecutive_connection_failures_before_delay
                     {
-                        dispatcher.delay_ready_queue();
+                        dispatcher.delay_ready_queue().await;
                     }
                     return Err(err);
                 }
@@ -583,7 +589,7 @@ impl Dispatcher {
         }
     }
 
-    fn throttle_ready_queue(&mut self, delay: Duration) {
+    async fn throttle_ready_queue(&mut self, delay: Duration) {
         let mut msgs: Vec<Message> = self.ready.lock().unwrap().drain(..).collect();
         if let Some(msg) = self.msg.take() {
             msgs.push(msg);
@@ -611,12 +617,13 @@ impl Dispatcher {
                     drop(activity);
                 })
             })
+            .await
             .expect("failed to spawn requeue");
         }
     }
 
     #[instrument(skip(self))]
-    fn bulk_ready_queue_operation(&mut self, response: Response) {
+    async fn bulk_ready_queue_operation(&mut self, response: Response) {
         let mut msgs: Vec<Message> = self.ready.lock().unwrap().drain(..).collect();
         if let Some(msg) = self.msg.take() {
             msgs.push(msg);
@@ -665,12 +672,13 @@ impl Dispatcher {
                     })
                 },
             )
+            .await
             .expect("bulk queue spawned");
         }
     }
 
     #[instrument(skip(self))]
-    fn delay_ready_queue(&mut self) {
+    async fn delay_ready_queue(&mut self) {
         tracing::debug!(
             "too many connection failures, delaying ready queue {}",
             self.name,
@@ -684,7 +692,8 @@ impl Dispatcher {
             }),
             content: "No answer from any hosts listed in MX".to_string(),
             command: None,
-        });
+        })
+        .await;
     }
 
     #[instrument(skip(self))]
@@ -749,7 +758,7 @@ impl Dispatcher {
 
                 if let Some(delay) = result.retry_after {
                     if delay >= self.path_config.client_timeouts.idle_timeout {
-                        self.throttle_ready_queue(delay);
+                        self.throttle_ready_queue(delay).await;
                         anyhow::bail!("connection rate throttled for {delay:?}");
                     }
                     tracing::trace!(
@@ -913,7 +922,7 @@ impl Dispatcher {
 
                 if let Some(delay) = result.retry_after {
                     if delay >= self.path_config.client_timeouts.idle_timeout {
-                        self.throttle_ready_queue(delay);
+                        self.throttle_ready_queue(delay).await;
                         return Ok(());
                     }
                     tracing::trace!("{} throttled message rate, sleep for {delay:?}", self.name);
@@ -959,7 +968,8 @@ impl Dispatcher {
                     .await;
                     rt_spawn("requeue message".to_string(), move || {
                         Ok(async move { Self::requeue_message(msg, true, None).await })
-                    })?;
+                    })
+                    .await?;
                 }
                 self.metrics.msgs_transfail.inc();
                 self.metrics.global_msgs_transfail.inc();
@@ -1032,7 +1042,7 @@ impl Drop for Dispatcher {
         // Ensure that we re-queue any message that we had popped
         if let Some(msg) = self.msg.take() {
             let activity = self.activity.clone();
-            rt_spawn("Dispatcher::drop".to_string(), move || {
+            rt_spawn_non_blocking("Dispatcher::drop".to_string(), move || {
                 Ok(async move {
                     if activity.is_shutting_down() {
                         Queue::save_if_needed_and_log(&msg).await;

crates/kumod/src/http_server/auth.rs

@@ -65,7 +65,8 @@ impl AuthKind {
         let (tx, rx) = tokio::sync::oneshot::channel();
         rt_spawn(format!("http auth validate {kind:?}"), move || {
             Ok(async move { tx.send(kind.validate_impl().await) })
-        })?;
+        })
+        .await?;
         rx.await?
     }
 

crates/kumod/src/http_server/inject_v1.rs

@@ -436,7 +436,8 @@ async fn process_recipient<'a>(
         .await;
         rt_spawn(format!("http inject for {peer_address:?}"), move || {
             Ok(async move { QueueManager::insert(&queue_name, message).await })
-        })?;
+        })
+        .await?;
     }
 
     Ok(())
@@ -497,7 +498,8 @@ pub async fn inject_v1(
     // Bounce to the thread pool where we can run async lua
     rt_spawn(format!("http inject_v1 for {peer_address:?}"), move || {
         Ok(async move { tx.send(inject_v1_impl(auth, sender, peer_address, request).await) })
-    })?;
+    })
+    .await?;
     rx.await?
 }
 

crates/kumod/src/lifecycle.rs

@@ -117,6 +117,7 @@ impl LifeCycle {
     /// LifeCycle instance to wake up and initiate the shutdown
     /// procedure.
     pub async fn request_shutdown() {
+        tracing::debug!("shutdown has been requested");
         if let Some(state) = STOPPING.get() {
             state.request_shutdown_tx.send(()).await.ok();
         }
@@ -132,6 +133,7 @@ impl LifeCycle {
             _ = tokio::signal::ctrl_c() => {}
             _ = self.request_shutdown_rx.recv() => {}
         };
+        tracing::debug!("wait_for_shutdown: shutdown requested!");
         println!("Shutdown requested, please wait while work is saved");
         // Signal that we are stopping
         tracing::debug!("Signal tasks that we are stopping");

crates/kumod/src/main.rs

@@ -51,6 +51,10 @@ struct Opt {
     #[arg(long, default_value = "full")]
     diag_format: DiagnosticFormat,
 
+    /// Whether to enable the diagnostic tokio console
+    #[arg(long)]
+    tokio_console: bool,
+
     /// If started as root, which user to run as.
     #[arg(long)]
     user: Option<String>,
@@ -145,7 +149,11 @@ async fn run(opts: Opt) -> anyhow::Result<()> {
     };
 
     tracing_subscriber::registry()
-        .with(console_subscriber::spawn())
+        .with(if opts.tokio_console {
+            Some(console_subscriber::spawn())
+        } else {
+            None
+        })
         .with(layer)
         .with(EnvFilter::from_env("KUMOD_LOG"))
         .with(metrics_tracing_context::MetricsLayer::new())
@@ -183,7 +191,8 @@ async fn run(opts: Opt) -> anyhow::Result<()> {
             tracing::info!("initialization complete");
             Ok::<(), anyhow::Error>(())
         })
-    })?;
+    })
+    .await?;
 
     life_cycle.wait_for_shutdown().await;
 

crates/kumod/src/queue.rs

@@ -382,7 +382,8 @@ impl Queue {
                     );
                 }
             })
-        })?;
+        })
+        .await?;
 
         Ok(handle)
     }
@@ -464,6 +465,7 @@ impl Queue {
 
     #[instrument(skip(self, msg))]
     async fn insert_delayed(&mut self, msg: Message) -> anyhow::Result<InsertResult> {
+        tracing::trace!("insert_delayed {}", msg.id());
         match self.queue.insert(Arc::new(msg.clone())) {
             Ok(_) => {
                 self.delayed_gauge.inc();
@@ -479,6 +481,7 @@ impl Queue {
 
     #[instrument(skip(self, msg))]
     async fn force_into_delayed(&mut self, msg: Message) -> anyhow::Result<()> {
+        tracing::trace!("force_into_delayed {}", msg.id());
         loop {
             msg.delay_with_jitter(60).await?;
             match self.insert_delayed(msg.clone()).await? {
@@ -492,6 +495,7 @@ impl Queue {
 
     #[instrument(skip(msg))]
     pub async fn save_if_needed(msg: &Message) -> anyhow::Result<()> {
+        tracing::trace!("save_if_needed {}", msg.id());
         if msg.needs_save() {
             msg.save().await?;
         }
@@ -499,7 +503,6 @@ impl Queue {
         Ok(())
     }
 
-    #[instrument(skip(msg))]
     pub async fn save_if_needed_and_log(msg: &Message) {
         if let Err(err) = Self::save_if_needed(msg).await {
             let id = msg.id();
@@ -513,6 +516,7 @@ impl Queue {
 
     #[instrument(skip(self, msg))]
     async fn insert_ready(&mut self, msg: Message) -> anyhow::Result<()> {
+        tracing::trace!("insert_ready {}", msg.id());
         match &self.queue_config.protocol {
             DeliveryProto::Smtp => {
                 let egress_source = self
@@ -557,6 +561,11 @@ impl Queue {
                 }
             }
             DeliveryProto::Maildir { maildir_path } => {
+                tracing::trace!(
+                    "Deliver msg {} to maildir at {}",
+                    msg.id(),
+                    maildir_path.display()
+                );
                 let maildir_path = maildir_path.to_path_buf();
 
                 msg.load_data_if_needed().await?;
@@ -623,10 +632,11 @@ impl Queue {
         }
     }
 
-    #[instrument(skip(self, msg))]
+    #[instrument(fields(self.name), skip(self, msg))]
     pub async fn insert(&mut self, msg: Message) -> anyhow::Result<()> {
         self.last_change = Instant::now();
 
+        tracing::trace!("insert msg {}", msg.id());
         if let Some(b) = AdminBounceEntry::get_for_queue_name(&self.name) {
             let id = *msg.id();
             b.log(msg, Some(&self.name)).await;
@@ -675,12 +685,15 @@ impl QueueManager {
     }
 
     /// Insert message into a queue named `name`.
+    #[instrument(skip(msg))]
     pub async fn insert(name: &str, msg: Message) -> anyhow::Result<()> {
+        tracing::trace!("QueueManager::insert");
         let entry = Self::resolve(name).await?;
         let mut entry = entry.lock().await;
         entry.insert(msg).await
     }
 
+    #[instrument]
     pub async fn resolve(name: &str) -> anyhow::Result<QueueHandle> {
         let mut mgr = MANAGER.lock().await;
         match mgr.named.get(name) {
@@ -733,7 +746,7 @@ async fn maintain_named_queue(queue: &QueueHandle) -> anyhow::Result<()> {
             }
 
             if q.activity.is_shutting_down() {
-                sleep_duration = Duration::from_secs(5);
+                sleep_duration = Duration::from_secs(1);
                 for msg in q.queue.drain() {
                     Queue::save_if_needed_and_log(&msg).await;
                     drop(msg);
@@ -757,6 +770,7 @@ async fn maintain_named_queue(queue: &QueueHandle) -> anyhow::Result<()> {
                 PopResult::Items(messages) => {
                     q.delayed_gauge.sub(messages.len() as i64);
                     let max_age = q.queue_config.get_max_age();
+                    tracing::trace!("{} msgs are now ready", messages.len());
 
                     for msg in messages {
                         let egress_source =

crates/kumod/src/runtime.rs

@@ -70,7 +70,7 @@ impl Runtime {
 ///
 /// If you are already in a !Send future, then using spawn_local below has
 /// less overhead.
-pub fn rt_spawn<F: (FnOnce() -> anyhow::Result<FUT>) + Send + 'static, FUT>(
+pub async fn rt_spawn<F: (FnOnce() -> anyhow::Result<FUT>) + Send + 'static, FUT>(
     name: String,
     func: F,
 ) -> anyhow::Result<JoinHandle<FUT::Output>>
@@ -81,9 +81,9 @@ where
     let (tx, rx) = bounded::<anyhow::Result<JoinHandle<FUT::Output>>>(1);
     RUNTIME
         .jobs
-        .send_blocking(Command::Run(Box::new(move || match (func)() {
+        .try_send(Command::Run(Box::new(move || match (func)() {
             Ok(future) => {
-                tx.send_blocking(
+                tx.try_send(
                     tokio::task::Builder::new()
                         .name(&name)
                         .spawn_local(future)
@@ -92,11 +92,33 @@ where
                 .ok();
             }
             Err(err) => {
-                tx.send_blocking(Err(err)).ok();
+                tx.try_send(Err(err)).ok();
             }
         })))
         .map_err(|err| anyhow::anyhow!("failed to send func to runtime thread: {err:#}"))?;
-    rx.recv_blocking()?
+    rx.recv().await?
+}
+
+pub fn rt_spawn_non_blocking<F: (FnOnce() -> anyhow::Result<FUT>) + Send + 'static, FUT>(
+    name: String,
+    func: F,
+) -> anyhow::Result<()>
+where
+    FUT: Future + 'static,
+{
+    RUNTIME
+        .jobs
+        .try_send(Command::Run(Box::new(move || match (func)() {
+            Ok(future) => {
+                if let Err(err) = tokio::task::Builder::new().name(&name).spawn_local(future) {
+                    tracing::error!("rt_spawn_non_blocking: error: {err:#}");
+                }
+            }
+            Err(err) => {
+                tracing::error!("rt_spawn_non_blocking: error: {err:#}");
+            }
+        })))
+        .map_err(|err| anyhow::anyhow!("failed to send func to runtime thread: {err:#}"))
 }
 
 /// Spawn a future as a task with a name.

crates/kumod/src/smtp_server.rs

@@ -220,7 +220,7 @@ impl EsmtpListenerParams {
                                 {
                                     tracing::error!("SmtpServer::run: {err:#}");
                             }
-                    }))?;
+                    })).await?;
                 }
             };
         }
@@ -299,7 +299,7 @@ impl RelayDisposition {
 }
 
 impl SmtpServer {
-    #[instrument]
+    #[instrument(skip(params, my_address, peer_address))]
     pub async fn run<T>(
         socket: T,
         my_address: SocketAddr,
@@ -385,7 +385,8 @@ impl SmtpServer {
         }
     }
 
-    async fn write_response<S: AsRef<str>>(
+    #[instrument(skip(self))]
+    async fn write_response<S: AsRef<str> + Debug>(
         &mut self,
         status: u16,
         message: S,
@@ -396,6 +397,7 @@ impl SmtpServer {
                 let is_last = lines.peek().is_none();
                 let sep = if is_last { ' ' } else { '-' };
                 let text = format!("{status}{sep}{line}\r\n");
+                tracing::trace!("writing response line: {text}");
                 socket
                     .write(text.as_bytes())
                     .await
@@ -414,8 +416,10 @@ impl SmtpServer {
         }
     }
 
+    #[instrument(skip(self))]
     async fn read_line(&mut self) -> anyhow::Result<ReadLine> {
         let mut too_long = false;
+        tracing::trace!("reading line");
         loop {
             let mut iter = self.read_buffer.iter().enumerate();
             while let Some((i, &b)) = iter.next() {