add reconnect_strategy option

This controls what we do in an smtp session when we experience a
disconnect during message sending; do we give up on the session,
continue with the connection plan, or try to connect to the same
host again and continue sending any additional messages there?

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(),
         }
     }
 }

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: [

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"
             );

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
 

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"`.
+