diff --git a/crates/integration-tests/maildir-sink.lua b/crates/integration-tests/maildir-sink.lua index 88569152..eec37b23 100644 --- a/crates/integration-tests/maildir-sink.lua +++ b/crates/integration-tests/maildir-sink.lua @@ -17,6 +17,9 @@ kumo.on('init', function() relay_hosts = { '0.0.0.0/0' }, batch_handling = 'BatchByDomain', max_recipients_per_message = 4, + -- This client_timeout value is coupled with assumptions + -- in disconnect_peer_idle_out! + client_timeout = '3s', } local client_ca = os.getenv 'KUMOD_CLIENT_REQUIRED_CA' if client_ca then diff --git a/crates/integration-tests/source.lua b/crates/integration-tests/source.lua index d6f03ebc..e5712bac 100644 --- a/crates/integration-tests/source.lua +++ b/crates/integration-tests/source.lua @@ -296,6 +296,21 @@ kumo.on('get_queue_config', function(domain, tenant, campaign, routing_domain) } end + -- This domain is coupled with the broken_first_choice_mx test + if domain == 'broken-first-mx.example.com' then + protocol = { + -- Redirect traffic to the sink, with two hosts in the connection plan, + -- but the first one is broken/not routable + smtp = { + mx_list = { + -- 255.255.255.255 should not be reachable/routable on any system + { name = 'brokenhost', addr = '255.255.255.255:' .. SINK_PORT }, + { name = 'workinghost', addr = '127.0.0.1:' .. SINK_PORT }, + }, + }, + } + end + if domain == 'nxdomain' then -- this nxdomain domain is a special domain that is assumed not -- to resolve. It is generated by the retry_schedule integration diff --git a/crates/integration-tests/src/test/broken_first_choice_mx.rs b/crates/integration-tests/src/test/broken_first_choice_mx.rs new file mode 100644 index 00000000..d4c095d7 --- /dev/null +++ b/crates/integration-tests/src/test/broken_first_choice_mx.rs @@ -0,0 +1,70 @@ +use crate::kumod::{DaemonWithMaildir, MailGenParams}; +use kumo_log_types::RecordType::Delivery; +use std::time::Duration; + +/// We're sending to broken-first-mx which is defined in source.lua +/// to have a non-routable first MX in its connection plan, followed +/// by the regular sink. This should cause a connection failure +/// for the first candidate, but we should then successfully +/// deliver to the second candidate. +#[tokio::test] +async fn broken_first_choice_mx() -> anyhow::Result<()> { + let mut daemon = DaemonWithMaildir::start().await?; + let mut client = daemon.smtp_client().await?; + + let response = MailGenParams { + recip: Some("first@broken-first-mx.example.com"), + ..Default::default() + } + .send(&mut client) + .await?; + eprintln!("{response:?}"); + anyhow::ensure!(response.code == 250); + + daemon + .wait_for_source_summary( + |summary| summary.get(&Delivery).copied().unwrap_or(0) >= 1, + Duration::from_secs(50), + ) + .await; + + daemon + .wait_for_maildir_count(1, Duration::from_secs(10)) + .await; + + daemon.stop_both().await?; + + let delivery_summary = daemon.dump_logs().await?; + k9::snapshot!( + delivery_summary, + " +DeliverySummary { + source_counts: { + Reception: 1, + Delivery: 1, + }, + sink_counts: { + Reception: 1, + Delivery: 1, + }, +} +" + ); + + k9::snapshot!( + daemon.source.accounting_stats()?, + " +AccountingStats { + received: 1, + delivered: 1, +} +" + ); + + let logs = daemon.source.collect_logs().await?; + let delivered = logs.iter().find(|record| record.kind == Delivery).unwrap(); + + k9::assert_equal!(delivered.peer_address.as_ref().unwrap().name, "workinghost"); + + Ok(()) +} diff --git a/crates/integration-tests/src/test/disconnect_peer_idle_out.rs b/crates/integration-tests/src/test/disconnect_peer_idle_out.rs new file mode 100644 index 00000000..a7328ec6 --- /dev/null +++ b/crates/integration-tests/src/test/disconnect_peer_idle_out.rs @@ -0,0 +1,87 @@ +use crate::kumod::{DaemonWithMaildir, MailGenParams}; +use kumo_log_types::RecordType::Delivery; +use std::time::Duration; + +#[tokio::test] +async fn disconnect_peer_idle_out() -> anyhow::Result<()> { + let mut daemon = DaemonWithMaildir::start().await?; + let mut client = daemon.smtp_client().await?; + + let response = MailGenParams { + recip: Some("first@example.com"), + ..Default::default() + } + .send(&mut client) + .await?; + eprintln!("{response:?}"); + anyhow::ensure!(response.code == 250); + + // Wait for the sink to idle out its connection. + // We're assuming that it is set to 3s, so we wait + // 4s to make things a little less racy + tokio::time::sleep(Duration::from_secs(4)).await; + + let response = MailGenParams { + recip: Some("second@example.com"), + ..Default::default() + } + .send(&mut client) + .await?; + eprintln!("{response:?}"); + anyhow::ensure!(response.code == 250); + + daemon + .wait_for_source_summary( + |summary| summary.get(&Delivery).copied().unwrap_or(0) >= 2, + Duration::from_secs(50), + ) + .await; + + daemon + .wait_for_maildir_count(2, Duration::from_secs(10)) + .await; + + daemon.stop_both().await?; + + let delivery_summary = daemon.dump_logs().await?; + k9::snapshot!( + delivery_summary, + " +DeliverySummary { + source_counts: { + Reception: 2, + Delivery: 2, + }, + sink_counts: { + Reception: 2, + Delivery: 2, + }, +} +" + ); + + k9::snapshot!( + daemon.source.accounting_stats()?, + " +AccountingStats { + received: 2, + delivered: 2, +} +" + ); + + let logs = daemon.source.collect_logs().await?; + let mut delivered_sessions = logs + .iter() + .filter_map(|record| match record.kind { + kumo_log_types::RecordType::Delivery => Some(record.session_id), + _ => None, + }) + .collect::>(); + + delivered_sessions.sort(); + delivered_sessions.dedup(); + assert_eq!(delivered_sessions.len(), 2, "two separate sessions"); + + Ok(()) +} diff --git a/crates/integration-tests/src/test/mod.rs b/crates/integration-tests/src/test/mod.rs index befd80a0..41e5a121 100644 --- a/crates/integration-tests/src/test/mod.rs +++ b/crates/integration-tests/src/test/mod.rs @@ -2,8 +2,10 @@ mod arc; mod auth_deliver; mod auth_deliver_invalid_password; +mod broken_first_choice_mx; mod disconnect_in_data; mod disconnect_in_mail_from; +mod disconnect_peer_idle_out; mod disconnect_reconnect_same_host; mod disconnect_terminate_ok; mod eightbitmime; diff --git a/crates/kumod/src/http_server/inject_v1.rs b/crates/kumod/src/http_server/inject_v1.rs index 405931fb..f875af0a 100644 --- a/crates/kumod/src/http_server/inject_v1.rs +++ b/crates/kumod/src/http_server/inject_v1.rs @@ -1,7 +1,7 @@ use crate::delivery_metrics::MetricsWrappedConnection; use crate::logging::disposition::{log_disposition, LogDisposition, RecordType}; use crate::queue::{DeliveryProto, QueueConfig, QueueManager}; -use crate::ready_queue::{Dispatcher, QueueDispatcher}; +use crate::ready_queue::{AttemptConnectionDisposition, Dispatcher, QueueDispatcher}; use crate::smtp_server::{default_hostname, TraceHeaders}; use crate::spool::SpoolManager; use anyhow::Context; @@ -1324,12 +1324,17 @@ impl QueueDispatcher for HttpInjectionGeneratorDispatcher { None => Ok(false), } } - async fn attempt_connection(&mut self, dispatcher: &mut Dispatcher) -> anyhow::Result<()> { + async fn attempt_connection( + &mut self, + dispatcher: &mut Dispatcher, + ) -> anyhow::Result { if self.connection.is_none() { self.connection .replace(dispatcher.metrics.wrap_connection(())); + Ok(AttemptConnectionDisposition::ConnectedNew) + } else { + Ok(AttemptConnectionDisposition::ReusedExisting) } - Ok(()) } async fn have_more_connection_candidates(&mut self, _dispatcher: &mut Dispatcher) -> bool { false diff --git a/crates/kumod/src/lua_deliver.rs b/crates/kumod/src/lua_deliver.rs index 1850610d..7151c55e 100644 --- a/crates/kumod/src/lua_deliver.rs +++ b/crates/kumod/src/lua_deliver.rs @@ -1,7 +1,7 @@ use crate::delivery_metrics::MetricsWrappedConnection; use crate::logging::disposition::{log_disposition, LogDisposition}; use crate::queue::{IncrementAttempts, InsertReason, QueueManager}; -use crate::ready_queue::{Dispatcher, QueueDispatcher}; +use crate::ready_queue::{AttemptConnectionDisposition, Dispatcher, QueueDispatcher}; use crate::smtp_server::RejectError; use crate::spool::SpoolManager; use anyhow::Context; @@ -133,9 +133,14 @@ impl QueueDispatcher for LuaQueueDispatcher { self.proto_config.max_batch_latency } - async fn attempt_connection(&mut self, dispatcher: &mut Dispatcher) -> anyhow::Result<()> { + async fn attempt_connection( + &mut self, + dispatcher: &mut Dispatcher, + ) -> anyhow::Result { match &self.connection { - ConnectionState::Connected(_) => return Ok(()), + ConnectionState::Connected(_) => { + return Ok(AttemptConnectionDisposition::ReusedExisting) + } ConnectionState::Disconnected => { anyhow::bail!("only one connection attempt per session"); } @@ -162,7 +167,7 @@ impl QueueDispatcher for LuaQueueDispatcher { self.connection = ConnectionState::Connected(connection_wrapper.map_connection(connection)); dispatcher.delivered_this_connection = 0; - Ok(()) + Ok(AttemptConnectionDisposition::ConnectedNew) } async fn have_more_connection_candidates(&mut self, _dispatcher: &mut Dispatcher) -> bool { diff --git a/crates/kumod/src/queue/insert_context.rs b/crates/kumod/src/queue/insert_context.rs index 3e48a2d4..c4688c40 100644 --- a/crates/kumod/src/queue/insert_context.rs +++ b/crates/kumod/src/queue/insert_context.rs @@ -79,6 +79,9 @@ pub enum InsertReason { /// The safey net in Dispatcher::Drop re-queued the message. /// This shouldn't happen; if you see this in a log, please report it! DispatcherDrop, + /// The peer unilaterally closed the connection before we started + /// delivery, so we want to try a new connection plan + PeerClosedConnection, } #[cfg(test)] diff --git a/crates/kumod/src/ready_queue.rs b/crates/kumod/src/ready_queue.rs index 0f685c20..103ce470 100644 --- a/crates/kumod/src/ready_queue.rs +++ b/crates/kumod/src/ready_queue.rs @@ -1180,6 +1180,14 @@ impl Drop for ReadyQueue { } } +#[derive(Copy, Clone, Debug)] +pub enum AttemptConnectionDisposition { + ConnectedNew, + ReusedExisting, + PeerClosedConnectionNeedNewSession, + PeerClosedConnectionContinueSession, +} + #[async_trait] pub trait QueueDispatcher: Debug + Send { async fn deliver_message( @@ -1188,7 +1196,10 @@ pub trait QueueDispatcher: Debug + Send { dispatcher: &mut Dispatcher, ) -> anyhow::Result<()>; - async fn attempt_connection(&mut self, dispatcher: &mut Dispatcher) -> anyhow::Result<()>; + async fn attempt_connection( + &mut self, + dispatcher: &mut Dispatcher, + ) -> anyhow::Result; async fn have_more_connection_candidates(&mut self, dispatcher: &mut Dispatcher) -> bool; async fn close_connection(&mut self, dispatcher: &mut Dispatcher) -> anyhow::Result; @@ -1431,78 +1442,109 @@ impl Dispatcher { } }; - if let Err(err) = result { - if OpportunisticInsecureTlsHandshakeError::is_match_anyhow(&err) { - num_opportunistic_tls_failures += 1; - } - connection_failures.push(format!("{err:#}")); - if !queue_dispatcher - .have_more_connection_candidates(&mut dispatcher) - .await - { - for msg in dispatcher.msgs.drain(..) { - let summary = if num_opportunistic_tls_failures == connection_failures.len() - { - "All failures are related to OpportunisticInsecure STARTTLS. \ + match result { + Err(err) => { + if OpportunisticInsecureTlsHandshakeError::is_match_anyhow(&err) { + num_opportunistic_tls_failures += 1; + } + connection_failures.push(format!("{err:#}")); + if !queue_dispatcher + .have_more_connection_candidates(&mut dispatcher) + .await + { + for msg in dispatcher.msgs.drain(..) { + let summary = + if num_opportunistic_tls_failures == connection_failures.len() { + "All failures are related to OpportunisticInsecure STARTTLS. \ Consider setting enable_tls=Disabled for this site. " - } else { - "" - }; + } else { + "" + }; - let response = Response { - code: 400, - enhanced_code: None, - content: format!( - "KumoMTA internal: \ + let response = Response { + code: 400, + enhanced_code: None, + content: format!( + "KumoMTA internal: \ failed to connect to any candidate \ hosts: {summary}{}", - connection_failures.join(", ") - ), - command: None, - }; + connection_failures.join(", ") + ), + command: None, + }; - log_disposition(LogDisposition { - kind: RecordType::TransientFailure, - msg: msg.clone(), - site: &dispatcher.name, - peer_address: None, - response: response.clone(), - egress_pool: Some(&dispatcher.egress_pool), - egress_source: Some(&dispatcher.egress_source.name), - relay_disposition: None, - delivery_protocol: Some(&dispatcher.delivery_protocol), - tls_info: None, - source_address: None, - provider: dispatcher.path_config.borrow().provider_name.as_deref(), - session_id: Some(dispatcher.session_id), - recipient_list: None, - }) - .await; - QueueManager::requeue_message( - msg, - IncrementAttempts::Yes, - None, - response, - InsertReason::LoggedTransientFailure.into(), - ) - .await?; - dispatcher.metrics.inc_transfail(); - } + log_disposition(LogDisposition { + kind: RecordType::TransientFailure, + msg: msg.clone(), + site: &dispatcher.name, + peer_address: None, + response: response.clone(), + egress_pool: Some(&dispatcher.egress_pool), + egress_source: Some(&dispatcher.egress_source.name), + relay_disposition: None, + delivery_protocol: Some(&dispatcher.delivery_protocol), + tls_info: None, + source_address: None, + provider: dispatcher.path_config.borrow().provider_name.as_deref(), + session_id: Some(dispatcher.session_id), + recipient_list: None, + }) + .await; + QueueManager::requeue_message( + msg, + IncrementAttempts::Yes, + None, + response, + InsertReason::LoggedTransientFailure.into(), + ) + .await?; + dispatcher.metrics.inc_transfail(); + } - if consecutive_connection_failures.fetch_add(1, Ordering::SeqCst) - > dispatcher - .path_config - .borrow() - .consecutive_connection_failures_before_delay - { - dispatcher.delay_ready_queue().await; + if consecutive_connection_failures.fetch_add(1, Ordering::SeqCst) + > dispatcher + .path_config + .borrow() + .consecutive_connection_failures_before_delay + { + dispatcher.delay_ready_queue().await; + } + dispatcher.release_leases().await; + return Err(err); } + tracing::debug!("{err:#}"); + // Try the next candidate MX address + continue; + } + Ok( + AttemptConnectionDisposition::ReusedExisting + | AttemptConnectionDisposition::ConnectedNew, + ) => { + // fall through to below logic to do the send + } + Ok(AttemptConnectionDisposition::PeerClosedConnectionNeedNewSession) => { + tracing::debug!( + "{} Peer closed the connection, will make new session", + dispatcher.name + ); dispatcher.release_leases().await; - return Err(err); + queue_dispatcher.close_connection(&mut dispatcher).await?; + // Push the message(s) batch into the ready queue so that they + // can be immediately tried in another session + dispatcher + .reinsert_ready_queue(InsertReason::PeerClosedConnection.into()) + .await; + return Ok(()); + } + Ok(AttemptConnectionDisposition::PeerClosedConnectionContinueSession) => { + tracing::debug!( + "{} Peer closed the connection, continue with session", + dispatcher.name + ); + // We don't have a current connection, so continue around + // the loop to try to open a new one + continue; } - tracing::debug!("{err:#}"); - // Try the next candidate MX address - continue; } connection_failures.clear(); diff --git a/crates/kumod/src/smtp_dispatcher.rs b/crates/kumod/src/smtp_dispatcher.rs index aad6beaa..77a58a09 100644 --- a/crates/kumod/src/smtp_dispatcher.rs +++ b/crates/kumod/src/smtp_dispatcher.rs @@ -4,7 +4,7 @@ use crate::http_server::admin_trace_smtp_client_v1::{ }; use crate::logging::disposition::{log_disposition, LogDisposition, RecordType}; use crate::queue::{IncrementAttempts, InsertReason, QueueManager, QueueState}; -use crate::ready_queue::{Dispatcher, QueueDispatcher}; +use crate::ready_queue::{AttemptConnectionDisposition, Dispatcher, QueueDispatcher}; use crate::smtp_server::ShuttingDownError; use crate::spool::SpoolManager; use anyhow::Context; @@ -294,7 +294,10 @@ impl SmtpDispatcher { })) } - async fn attempt_connection_impl(&mut self, dispatcher: &mut Dispatcher) -> anyhow::Result<()> { + async fn attempt_connection_impl( + &mut self, + dispatcher: &mut Dispatcher, + ) -> anyhow::Result { if let Some(client) = &mut self.client { if client.is_connected() { // If we get a unilateral response here now it can either be: @@ -311,23 +314,64 @@ impl SmtpDispatcher { match client.check_unilateral_response().await { Ok(None) => { // Still connected - return Ok(()); + return Ok(AttemptConnectionDisposition::ReusedExisting); } Ok(Some(response)) => { + // We got an explicit signal that the connection is closing + // out. We don't consider this to be a transport error + // so we'll close the current connection. + // If we've managed to send messages so far in this session, + // then we're consider it to be a successful plan so we should + // not advance to the next candidate in the plan and make a + // new session. + // However, if we haven't managed to send anything so far, + // closing and restarting the plan now would likely lead to + // a persistent recurrence of this same state, so in that + // situation we need to ensure that the reconnect_strategy + // is applied to decide what to do. tracing::debug!( "{} sent a unilateral response: \ {response:?}, treating the connection as closed", dispatcher.name ); - let _ = self.close_connection(dispatcher).await; + // Decide whether this is a successful plan + self.update_state_for_reconnect(dispatcher); + + // if so, we can/should close the current session and start + // a new one with a fresh plan + if self.terminated_ok { + let _ = self.close_connection(dispatcher).await; + return Ok( + AttemptConnectionDisposition::PeerClosedConnectionNeedNewSession, + ); + } + + // otherwise, continue to next candidate host if that is what + // the reconnect_strategy indicates. + // While this is a return that leaves this function, + // our caller will typically continue its loop and call + // back in, but on the next call we won't show as connected + // and will instead reach the logic below to make a new + // connection in the current session, if that is what + // the reconnect_strategy indicated. + return Ok( + AttemptConnectionDisposition::PeerClosedConnectionContinueSession, + ); } Err(err) => { + // We got a transport error of some kind. + // update_state_for_reconnect applies any reconnect_strategy + // which will decide whether we continue with the connection + // plan or whether we need to go to a new session. tracing::debug!( "{} had error: {err:#} \ while checking for liveness, treating it as closed", dispatcher.name ); - self.client.take(); + self.update_state_for_reconnect(dispatcher); + return Ok( + AttemptConnectionDisposition::PeerClosedConnectionContinueSession, + ); } }; } @@ -845,7 +889,7 @@ impl SmtpDispatcher { .replace(connection_wrapper.map_connection(client)); self.client_address.replace(address); dispatcher.delivered_this_connection = 0; - Ok(()) + Ok(AttemptConnectionDisposition::ConnectedNew) } async fn resolve_cached_client_cert( @@ -971,7 +1015,10 @@ impl QueueDispatcher for SmtpDispatcher { self.terminated_ok } - async fn attempt_connection(&mut self, dispatcher: &mut Dispatcher) -> anyhow::Result<()> { + async fn attempt_connection( + &mut self, + dispatcher: &mut Dispatcher, + ) -> anyhow::Result { self.attempt_connection_impl(dispatcher) .await .map_err(|err| { diff --git a/crates/kumod/src/smtp_server.rs b/crates/kumod/src/smtp_server.rs index ecfab6ec..d39a3ceb 100644 --- a/crates/kumod/src/smtp_server.rs +++ b/crates/kumod/src/smtp_server.rs @@ -6,7 +6,7 @@ use crate::logging::disposition::{log_disposition, LogDisposition, RecordType}; use crate::logging::rejection::{log_rejection, LogRejection}; use crate::metrics_helper::smtp_rejected_for_service; use crate::queue::{DeliveryProto, IncrementAttempts, InsertReason, QueueConfig, QueueManager}; -use crate::ready_queue::{Dispatcher, QueueDispatcher}; +use crate::ready_queue::{AttemptConnectionDisposition, Dispatcher, QueueDispatcher}; use crate::spool::SpoolManager; use anyhow::{anyhow, Context}; use async_trait::async_trait; @@ -3432,12 +3432,17 @@ impl QueueDispatcher for DeferredSmtpInjectionDispatcher { } } - async fn attempt_connection(&mut self, dispatcher: &mut Dispatcher) -> anyhow::Result<()> { + async fn attempt_connection( + &mut self, + dispatcher: &mut Dispatcher, + ) -> anyhow::Result { if self.connection.is_none() { self.connection .replace(dispatcher.metrics.wrap_connection(())); + Ok(AttemptConnectionDisposition::ConnectedNew) + } else { + Ok(AttemptConnectionDisposition::ReusedExisting) } - Ok(()) } async fn have_more_connection_candidates(&mut self, _dispatcher: &mut Dispatcher) -> bool { diff --git a/crates/kumod/src/xfer/mod.rs b/crates/kumod/src/xfer/mod.rs index f1be663a..1f2f418d 100644 --- a/crates/kumod/src/xfer/mod.rs +++ b/crates/kumod/src/xfer/mod.rs @@ -1,7 +1,7 @@ use crate::http_server::inject_v1::activity_for_peer; use crate::logging::disposition::{log_disposition, LogDisposition}; use crate::queue::{DeliveryProto, QueueConfig, QueueManager}; -use crate::ready_queue::{Dispatcher, QueueDispatcher}; +use crate::ready_queue::{AttemptConnectionDisposition, Dispatcher, QueueDispatcher}; use crate::spool::SpoolManager; use anyhow::Context; use async_trait::async_trait; @@ -149,8 +149,11 @@ impl QueueDispatcher for XferDispatcher { Ok(true) } - async fn attempt_connection(&mut self, _dispatcher: &mut Dispatcher) -> anyhow::Result<()> { - Ok(()) + async fn attempt_connection( + &mut self, + _dispatcher: &mut Dispatcher, + ) -> anyhow::Result { + Ok(AttemptConnectionDisposition::ReusedExisting) } async fn have_more_connection_candidates(&mut self, _dispatcher: &mut Dispatcher) -> bool {