diff --git a/crates/integration-tests/src/kumod.rs b/crates/integration-tests/src/kumod.rs index 01e344a3..fef987d6 100644 --- a/crates/integration-tests/src/kumod.rs +++ b/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) } diff --git a/crates/integration-tests/src/main.rs b/crates/integration-tests/src/main.rs index 3d612731..1b2fbf07 100644 --- a/crates/integration-tests/src/main.rs +++ b/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: \r\n\ To: \r\n\ Subject: a test message\r\n\ diff --git a/crates/kumod/src/egress_path.rs b/crates/kumod/src/egress_path.rs index 45c63093..8274bba2 100644 --- a/crates/kumod/src/egress_path.rs +++ b/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 = 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 = 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; diff --git a/crates/kumod/src/http_server/auth.rs b/crates/kumod/src/http_server/auth.rs index dac9ece2..28df56d4 100644 --- a/crates/kumod/src/http_server/auth.rs +++ b/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? } diff --git a/crates/kumod/src/http_server/inject_v1.rs b/crates/kumod/src/http_server/inject_v1.rs index 5ddd0812..5fdfbfbc 100644 --- a/crates/kumod/src/http_server/inject_v1.rs +++ b/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? } diff --git a/crates/kumod/src/lifecycle.rs b/crates/kumod/src/lifecycle.rs index a606cb03..06e527c8 100644 --- a/crates/kumod/src/lifecycle.rs +++ b/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"); diff --git a/crates/kumod/src/main.rs b/crates/kumod/src/main.rs index cda169fb..6358ad0a 100644 --- a/crates/kumod/src/main.rs +++ b/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, @@ -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; diff --git a/crates/kumod/src/queue.rs b/crates/kumod/src/queue.rs index c8a028f2..ae083767 100644 --- a/crates/kumod/src/queue.rs +++ b/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 { + 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 { 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 = diff --git a/crates/kumod/src/runtime.rs b/crates/kumod/src/runtime.rs index c095df92..ccf06ec0 100644 --- a/crates/kumod/src/runtime.rs +++ b/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 anyhow::Result) + Send + 'static, FUT>( +pub async fn rt_spawn anyhow::Result) + Send + 'static, FUT>( name: String, func: F, ) -> anyhow::Result> @@ -81,9 +81,9 @@ where let (tx, rx) = bounded::>>(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 anyhow::Result) + 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. diff --git a/crates/kumod/src/smtp_server.rs b/crates/kumod/src/smtp_server.rs index 5c2a5965..0ad8f8d9 100644 --- a/crates/kumod/src/smtp_server.rs +++ b/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( socket: T, my_address: SocketAddr, @@ -385,7 +385,8 @@ impl SmtpServer { } } - async fn write_response>( + #[instrument(skip(self))] + async fn write_response + 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 { 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() {