fix delayed shutdown for non-smtp delivery handlers

Occasionally we'll have someone report that systemd timed out
and sigkill'd their kumo on shutdown.

One possible scenario for this is a lua delivery handler that
is taking too long, presumably because the other end of it
(eg: webhook or other custom endpoint) is not responding in
a timely fashion.

The way that we handle shutdown is that we compute a maximum
theoretical timeout value by summing up all of the smtp client
timeout values.  Some of those can be several minutes in
duration because the are using default values derived from
a very conservative set of values suggested by the SMTP
RFCs from the '70s.

Those obviously should not apply to a custom delivery handler,
but also, in the context of an established SMTP session, we
should not add in the connection-establishment-specific values
when we're just waiting for a per-message send.

This commit addresses this situation on two fronts:

* Introduce a new system_shutdown_timeout value that allows the
  user to conveniently express their desired timeout value
  in a single option.  This is *not* set by default!
* The default value for system_shutdown_timeout is computed by
  summing the per-message-delivery smtp timeout options, which
  is a much more reasonable, and more importantly, shorter than
  our 300s TimeoutStopSec value in kumomta.service

Author: wez

assets/policy-extras/shaping.toml

@@ -232,3 +232,4 @@ connection_limit = "local:5"
 # Allow a healthy amount in the ready queue in case things get
 # busy
 max_ready = 10000
+system_shutdown_timeout = "30s"

crates/kumo-api-types/src/egress_path.rs

@@ -215,6 +215,12 @@ pub struct EgressPathConfig {
     #[serde(flatten)]
     pub client_timeouts: SmtpClientTimeouts,
 
+    /// How long to wait for an established session to gracefully
+    /// close when the system is shutting down. After this period
+    /// has elapsed, sessions will be aborted.
+    #[serde(default, with = "duration_serde")]
+    pub system_shutdown_timeout: Option<Duration>,
+
     #[serde(default = "EgressPathConfig::default_max_ready")]
     pub max_ready: usize,
 
@@ -342,6 +348,7 @@ impl Default for EgressPathConfig {
             max_connection_rate: None,
             max_deliveries_per_connection: Self::default_max_deliveries_per_connection(),
             client_timeouts: SmtpClientTimeouts::default(),
+            system_shutdown_timeout: None,
             prohibited_hosts: CidrSet::default_prohibited_hosts(),
             skip_hosts: CidrSet::default(),
             ehlo_domain: None,

crates/kumo-api-types/src/shaping.rs

@@ -1948,6 +1948,7 @@ MergedEntry {
             starttls_timeout: 5s,
             auth_timeout: 60s,
         },
+        system_shutdown_timeout: None,
         max_ready: 1024,
         consecutive_connection_failures_before_delay: 100,
         smtp_port: 25,
@@ -2092,6 +2093,7 @@ MergedEntry {
             starttls_timeout: 5s,
             auth_timeout: 60s,
         },
+        system_shutdown_timeout: None,
         max_ready: 1024,
         consecutive_connection_failures_before_delay: 100,
         smtp_port: 25,
@@ -2153,6 +2155,7 @@ MergedEntry {
                 starttls_timeout: 5s,
                 auth_timeout: 60s,
             },
+            system_shutdown_timeout: None,
             max_ready: 1024,
             consecutive_connection_failures_before_delay: 100,
             smtp_port: 25,
@@ -2299,6 +2302,7 @@ MergedEntry {
             starttls_timeout: 5s,
             auth_timeout: 60s,
         },
+        system_shutdown_timeout: None,
         max_ready: 1024,
         consecutive_connection_failures_before_delay: 100,
         smtp_port: 25,

crates/kumod/src/ready_queue.rs

@@ -571,7 +571,11 @@ impl ReadyQueueManager {
                 _ = wait_for_shutdown => {
                     shutting_down = true;
                     if force_reap_deadline.is_none() {
-                        let duration = queue.path_config.borrow().client_timeouts.total_message_send_duration();
+                        let path_config = queue.path_config.borrow();
+
+                        let duration = path_config.system_shutdown_timeout.unwrap_or_else(|| {
+                            path_config.client_timeouts.total_message_send_duration()
+                        });
                         force_reap_deadline.replace(tokio::time::Instant::now() + duration);
                         tracing::debug!("{name}: reap deadline in {duration:?}");
                     }
@@ -918,7 +922,9 @@ impl ReadyQueue {
             return;
         }
 
-        let lease_duration = path_config.client_timeouts.total_message_send_duration();
+        let lease_duration = path_config
+            .client_timeouts
+            .total_connection_lifetime_duration();
         let limit_name = format!("kumomta.connection_limit.{}", self.name);
         let mut limits = vec![(
             &limit_name,
@@ -1931,7 +1937,7 @@ impl Dispatcher {
                     self.path_config
                         .borrow()
                         .client_timeouts
-                        .total_message_send_duration(),
+                        .total_connection_lifetime_duration(),
                 )
                 .await
                 .is_err()

crates/rfc5321/src/client_types.rs

@@ -138,8 +138,9 @@ impl SmtpClientTimeouts {
         }
     }
 
-    /// Compute theoretical maximum lifetime of a single message send
-    pub fn total_message_send_duration(&self) -> Duration {
+    /// Compute theoretical maximum lifetime of a connection when
+    /// sending a single message
+    pub fn total_connection_lifetime_duration(&self) -> Duration {
         self.connect_timeout
             + self.banner_timeout
             + self.ehlo_timeout
@@ -151,6 +152,15 @@ impl SmtpClientTimeouts {
             + self.starttls_timeout
             + self.idle_timeout
     }
+
+    /// Compute theoretical maximum lifetime of a single message send
+    /// on an already established connection
+    pub fn total_message_send_duration(&self) -> Duration {
+            self.mail_from_timeout
+            + self.rcpt_to_timeout
+            + self.data_timeout
+            + self.data_dot_timeout
+    }
 }
 
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone, Hash)]

docs/changelog/main.md

@@ -47,6 +47,11 @@
   version, you may need to review and adjust both that and the corresponding
   [kumo.set_ready_qmaint_threads](../reference/kumo/set_ready_qmaint_threads.md)
   tuning.
+* New
+  [system_shutdown_timeout](../reference/kumo/make_egress_path/system_shutdown_timeout.md)
+  option to specify how long we should wait for an in-flight delivery attempt
+  to wrap up before terminating it once we have received a request to shutdown
+  kumod.
 
 ## Fixes
 
@@ -56,3 +61,6 @@
 * tsa-daemon now increases its soft `NOFILE` limit to match its hard limit
   on startup (just as we do in kumod), which helps to avoid issues with
   running out of file descriptors on systems with very large core counts.
+* Shutdown could take longer than the 300s permitted by kumomta.service
+  when lua delivery handlers are experiencing delays, leading to systemd
+  issuing a SIGKILL.

docs/reference/kumo/make_egress_path/system_shutdown_timeout.md

@@ -0,0 +1,22 @@
+# system_shutdown_timeout
+
+{{since('dev')}}
+
+How long to wait for an in-flight delivery attempt to gracefully complete when
+the system is being shutdown. Once this timeout is reached, any open sessions
+will be aborted.
+
+The default value is computed by summing up the per-message-delivery timeout
+values:
+
+* [mail_from_timeout](mail_from_timeout.md) +
+* [rcpt_to_timeout](rcpt_to_timeout.md) +
+* [data_timeout](data_timeout.md) +
+* [data_dot_timeout](data_dot_timeout.md)
+
+!!! caution
+    If you make this value too short you increase the risk duplicate delivery
+    when the sessions are terminated.
+
+
+