diff --git a/crates/kumo-api-types/src/egress_path.rs b/crates/kumo-api-types/src/egress_path.rs index 8aa27f91..25222fdd 100644 --- a/crates/kumo-api-types/src/egress_path.rs +++ b/crates/kumo-api-types/src/egress_path.rs @@ -134,6 +134,22 @@ pub enum ConfigRefreshStrategy { Epoch, } +#[derive(Deserialize, Serialize, Debug, Clone, Default, Copy, PartialEq, Eq)] +#[cfg_attr(feature = "lua", derive(FromLua))] +pub enum ReconnectStrategy { + /// Close out the current connection session, allowing the maintainer + /// to decide about opening a new session and starting with a fresh + /// connection plan + TerminateSession, + /// Try to reconnect to the same host that we were using and where + /// we experienced the error + ReconnectSameHost, + /// Advance to the next host in the connection, if any. If none remain, + /// this is equivalent to TerminateSession + #[default] + ConnectNextHost, +} + #[derive(Deserialize, Serialize, Debug, Clone, PartialEq)] #[cfg_attr(feature = "lua", derive(FromLua))] #[serde(deny_unknown_fields)] @@ -260,6 +276,11 @@ pub struct EgressPathConfig { /// If true, rather than ESMTP, use the LMTP protocol #[serde(default)] pub use_lmtp: bool, + + /// How to behave if we experience either a 421 response, an IO Error, + /// or a timeout while talking to the peer. + #[serde(default)] + pub reconnect_strategy: ReconnectStrategy, } #[cfg(feature = "lua")] @@ -306,6 +327,7 @@ impl Default for EgressPathConfig { remember_broken_tls: None, opportunistic_tls_reconnect_on_failed_handshake: false, use_lmtp: false, + reconnect_strategy: ReconnectStrategy::default(), } } } diff --git a/crates/kumo-api-types/src/shaping.rs b/crates/kumo-api-types/src/shaping.rs index e26ae407..ee3006c3 100644 --- a/crates/kumo-api-types/src/shaping.rs +++ b/crates/kumo-api-types/src/shaping.rs @@ -1906,6 +1906,7 @@ MergedEntry { remember_broken_tls: None, opportunistic_tls_reconnect_on_failed_handshake: false, use_lmtp: false, + reconnect_strategy: ConnectNextHost, }, sources: {}, automation: [ @@ -2044,6 +2045,7 @@ MergedEntry { remember_broken_tls: None, opportunistic_tls_reconnect_on_failed_handshake: false, use_lmtp: false, + reconnect_strategy: ConnectNextHost, }, sources: { "my source name": EgressPathConfig { @@ -2095,6 +2097,7 @@ MergedEntry { remember_broken_tls: None, opportunistic_tls_reconnect_on_failed_handshake: false, use_lmtp: false, + reconnect_strategy: ConnectNextHost, }, }, automation: [ @@ -2239,6 +2242,7 @@ MergedEntry { remember_broken_tls: None, opportunistic_tls_reconnect_on_failed_handshake: false, use_lmtp: false, + reconnect_strategy: ConnectNextHost, }, sources: {}, automation: [ diff --git a/crates/kumod/src/smtp_dispatcher.rs b/crates/kumod/src/smtp_dispatcher.rs index 572c3b4a..cddacef2 100644 --- a/crates/kumod/src/smtp_dispatcher.rs +++ b/crates/kumod/src/smtp_dispatcher.rs @@ -11,7 +11,7 @@ use async_trait::async_trait; use config::{load_config, CallbackSignature}; use dns_resolver::{resolve_a_or_aaaa, ResolvedMxAddresses}; use kumo_address::socket::SocketAddress; -use kumo_api_types::egress_path::{EgressPathConfig, Tls}; +use kumo_api_types::egress_path::{EgressPathConfig, ReconnectStrategy, Tls}; use kumo_log_types::{MaybeProxiedSourceAddress, ResolvedAddress}; use kumo_server_lifecycle::ShutdownSubcription; use kumo_server_runtime::spawn; @@ -781,6 +781,23 @@ impl SmtpDispatcher { }) .await } + + fn update_state_for_reconnect(&mut self, dispatcher: &mut Dispatcher) { + match dispatcher.path_config.borrow().reconnect_strategy { + ReconnectStrategy::TerminateSession => { + self.addresses.clear(); + } + ReconnectStrategy::ReconnectSameHost => { + if let Some(address) = self.client_address.take() { + self.addresses.push(address); + } + } + ReconnectStrategy::ConnectNextHost => { + // Nothing needed; we're naturally set up to do this + } + } + self.client.take(); + } } #[async_trait] @@ -959,6 +976,12 @@ impl QueueDispatcher for SmtpDispatcher { dispatcher.name, self.client_address ); + if response.code == 421 { + // We're effectively disconnected, so prepare + // for reconnecting for the next message. + self.update_state_for_reconnect(dispatcher); + } + if let Some(msg) = dispatcher.msgs.pop() { self.log_disposition( dispatcher, @@ -1047,6 +1070,7 @@ impl QueueDispatcher for SmtpDispatcher { } dispatcher.metrics.inc_transfail(); // Move on to the next host + self.update_state_for_reconnect(dispatcher); anyhow::bail!("{reason}"); } Err(ClientError::TimeOutResponse { command, duration }) => { @@ -1089,10 +1113,12 @@ impl QueueDispatcher for SmtpDispatcher { } dispatcher.metrics.inc_transfail(); // Move on to the next host + self.update_state_for_reconnect(dispatcher); anyhow::bail!("{reason}"); } Err(err) => { // Transient failure; continue with another host + self.update_state_for_reconnect(dispatcher); tracing::debug!( "failed to send message to {} {:?}: {err:#}", dispatcher.name, @@ -1117,6 +1143,7 @@ impl QueueDispatcher for SmtpDispatcher { .map(|c| c.is_connected()) .unwrap_or(false); if !is_connected { + self.update_state_for_reconnect(dispatcher); anyhow::bail!( "after previous send attempt, client is unexpectedly no longer connected" ); diff --git a/docs/changelog/main.md b/docs/changelog/main.md index da7f02db..4c3da934 100644 --- a/docs/changelog/main.md +++ b/docs/changelog/main.md @@ -5,6 +5,10 @@ ## Other Changes and Enhancements * Added [kcli inspect-sched-q](../reference/kcli/inspect-sched-q.md) command. #231 +* Added + [reconnect_strategy](../reference/kumo/make_egress_path/reconnect_strategy.md) + egress path option to control what happens with a session that experiences + a disconnection during message sending. ## Fixes diff --git a/docs/reference/kumo/make_egress_path/reconnect_strategy.md b/docs/reference/kumo/make_egress_path/reconnect_strategy.md new file mode 100644 index 00000000..5fa51316 --- /dev/null +++ b/docs/reference/kumo/make_egress_path/reconnect_strategy.md @@ -0,0 +1,40 @@ +# reconnect_strategy + +{{since('dev')}} + +Controls the behavior of the SMTP dispatcher when it encounters an error during +message delivery. It does **not** affect connection-time errors (such as connection +failure, or protocol or transport errors around banner, EHLO, STARTTLS, AUTH), +and specifically only targets errors that might arise as part of delivering +a message: + +* A 421 protocol response where the peer closes the connection. +* A timeout writing a request +* A timeout reading a response +* Some other IO error on the transport (eg: connection reset) + +You can set the `reconnect_strategy` to one of the following values to select +the desired behavior for session re-use for subsequent messages: + +* `"TerminateSession"` - close the current session, allowing the queue + maintainer to decide about opening a new connection based on your shaping + configuration. If a new session is established, it will start with a fresh + connection plan. +* `"ReconnectSameHost"` - close the current connection, but adjust the session + state so that it will try connecting to the same host again for future sends. +* `"ConnectNextHost"` - close the current connection and proceed to the next + host in the connection plan. This is the default behavior. + +The connection plan is constructed when a session is initiated; it is drawn +from the preference-ordered list of MX hosts, but randomizes the set of hosts +at each preference level. This is then flattened into a list of hosts that +will be attempted one after the other to establish a connection. + +`"ConnectNextHost"` will maximize the chances of delivering mail in the face of +various transient issues with the destination site. + +Some sites have very opininated anti-abuse policies and consider any attempt to +connect to second tier (non-preferred) MX hosts as signs of bad behavior and +this may impact your effective deliverability. For those sites you may want to +consider deploying with `reconnect_strategy="TerminateSession"`. +