add (low|no)_memory_reduction_policy options

These provide more control over the memory vs spool IO tradeoff
when memory is short.

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

@@ -126,6 +126,15 @@ pub fn find_rustls_cipher_suite(name: &str) -> Option<SupportedCipherSuite> {
     None
 }
 
+#[derive(Deserialize, Serialize, Debug, Clone, Default, Copy, PartialEq, Eq)]
+#[cfg_attr(feature = "lua", derive(FromLua))]
+pub enum MemoryReductionPolicy {
+    #[default]
+    ShrinkDataAndMeta,
+    ShrinkData,
+    NoShrink,
+}
+
 #[derive(Deserialize, Serialize, Debug, Clone, Default, Copy, PartialEq, Eq)]
 #[cfg_attr(feature = "lua", derive(FromLua))]
 pub enum ConfigRefreshStrategy {
@@ -285,6 +294,14 @@ pub struct EgressPathConfig {
     /// Which thread pool to use for processing the ready queue
     #[serde(default)]
     pub readyq_pool_name: Option<String>,
+
+    /// What to do to newly inserted messages when memory is low
+    #[serde(default)]
+    pub low_memory_reduction_policy: MemoryReductionPolicy,
+
+    /// What to do to newly inserted messages when memory is over the soft limit
+    #[serde(default)]
+    pub no_memory_reduction_policy: MemoryReductionPolicy,
 }
 
 #[cfg(feature = "lua")]
@@ -333,6 +350,8 @@ impl Default for EgressPathConfig {
             use_lmtp: false,
             reconnect_strategy: ReconnectStrategy::default(),
             readyq_pool_name: None,
+            low_memory_reduction_policy: MemoryReductionPolicy::default(),
+            no_memory_reduction_policy: MemoryReductionPolicy::default(),
         }
     }
 }

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

@@ -1913,6 +1913,8 @@ MergedEntry {
         use_lmtp: false,
         reconnect_strategy: ConnectNextHost,
         readyq_pool_name: None,
+        low_memory_reduction_policy: ShrinkDataAndMeta,
+        no_memory_reduction_policy: ShrinkDataAndMeta,
     },
     sources: {},
     automation: [
@@ -2053,6 +2055,8 @@ MergedEntry {
         use_lmtp: false,
         reconnect_strategy: ConnectNextHost,
         readyq_pool_name: None,
+        low_memory_reduction_policy: ShrinkDataAndMeta,
+        no_memory_reduction_policy: ShrinkDataAndMeta,
     },
     sources: {
         "my source name": EgressPathConfig {
@@ -2106,6 +2110,8 @@ MergedEntry {
             use_lmtp: false,
             reconnect_strategy: ConnectNextHost,
             readyq_pool_name: None,
+            low_memory_reduction_policy: ShrinkDataAndMeta,
+            no_memory_reduction_policy: ShrinkDataAndMeta,
         },
     },
     automation: [
@@ -2252,6 +2258,8 @@ MergedEntry {
         use_lmtp: false,
         reconnect_strategy: ConnectNextHost,
         readyq_pool_name: None,
+        low_memory_reduction_policy: ShrinkDataAndMeta,
+        no_memory_reduction_policy: ShrinkDataAndMeta,
     },
     sources: {},
     automation: [

crates/kumo-server-memory/src/lib.rs

@@ -486,6 +486,24 @@ pub fn low_memory() -> bool {
     LOW_MEM.load(Ordering::SeqCst)
 }
 
+/// Indicates the overall memory status
+pub fn memory_status() -> MemoryStatus {
+    if get_headroom() == 0 {
+        MemoryStatus::NoMemory
+    } else if low_memory() {
+        MemoryStatus::LowMemory
+    } else {
+        MemoryStatus::Ok
+    }
+}
+
+#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+pub enum MemoryStatus {
+    Ok,
+    LowMemory,
+    NoMemory,
+}
+
 /// Returns a receiver that will notify when memory status
 /// changes from OK -> !OK or vice versa.
 pub fn subscribe_to_memory_status_changes() -> Option<Receiver<()>> {

crates/kumod/src/ready_queue.rs

@@ -21,10 +21,12 @@ use async_trait::async_trait;
 use config::epoch::ConfigEpoch;
 use config::{load_config, CallbackSignature};
 use dns_resolver::MailExchanger;
-use kumo_api_types::egress_path::{ConfigRefreshStrategy, EgressPathConfig};
+use kumo_api_types::egress_path::{ConfigRefreshStrategy, EgressPathConfig, MemoryReductionPolicy};
 use kumo_server_common::config_handle::ConfigHandle;
 use kumo_server_lifecycle::{is_shutting_down, Activity, ShutdownSubcription};
-use kumo_server_memory::{get_headroom, low_memory, subscribe_to_memory_status_changes_async};
+use kumo_server_memory::{
+    get_headroom, memory_status, subscribe_to_memory_status_changes_async, MemoryStatus,
+};
 use kumo_server_runtime::{get_named_runtime, spawn, Runtime};
 use message::message::{MessageList, QueueNameComponents};
 use message::Message;
@@ -614,8 +616,19 @@ impl ReadyQueue {
 
     pub async fn insert(&self, msg: Message) -> Result<(), Message> {
         let _timer = INSERT_LATENCY.start_timer();
-        if low_memory() {
-            msg.save_and_shrink().await.ok();
+        let action = match memory_status() {
+            MemoryStatus::LowMemory => self.path_config.borrow().low_memory_reduction_policy,
+            MemoryStatus::NoMemory => self.path_config.borrow().no_memory_reduction_policy,
+            MemoryStatus::Ok => MemoryReductionPolicy::NoShrink,
+        };
+        match action {
+            MemoryReductionPolicy::NoShrink => {}
+            MemoryReductionPolicy::ShrinkDataAndMeta => {
+                msg.save_and_shrink().await.ok();
+            }
+            MemoryReductionPolicy::ShrinkData => {
+                msg.save_and_shrink_data().await.ok();
+            }
         }
         match self.ready.push(msg) {
             Ok(()) => {

docs/changelog/main.md

@@ -22,6 +22,10 @@
   for a list of the additional functions.
 * New [kumo.string.eval_template](../reference/string/eval_template.md)
   function for expanding minijinja template strings.
+* New [low_memory_reduction_policy](../reference/kumo/make_egress_path/low_memory_reduction_policy.md) and
+  [no_memory_reduction_policy](../reference/kumo/make_egress_path/no_memory_reduction_policy.md)
+  options give advanced control over memory vs. spool IO trade-offs when
+  available is memory low.
 
 ## Fixes
 

docs/reference/kumo/make_egress_path/low_memory_reduction_policy.md

@@ -0,0 +1,36 @@
+---
+tags:
+ - memory
+---
+
+# low_memory_reduction_policy
+
+{{since('dev')}}
+
+Specifies what action should be taken when a message is added to the ready
+queue corresponding to this egress path when the memory usage is above the *low
+memory threshold*.
+
+Possible values for this option are:
+
+* `"ShrinkDataAndMeta"` - this is the default (and is the implicit action for
+  older versions of KumoMTA).  Both the message data and metadata will be saved
+  (if modified since the prior save, or if the message has not yet been saved
+  to spool), then both will be released, freeing up the corresponding memory.
+* `"ShrinkData"` - Both the message data and metadata will be saved
+  (if modified since the prior save, or if the message has not yet been saved
+  to spool), then just the message data will be released, freeing up that memory.
+  The metadata will be preserved.
+* `"NoShrink"` - do not save or free up any message memory.
+
+This setting allows you more control in the trade-off of memory usage against
+spool IO pressure. The default is relatively conservative, aiming to avoid OOM
+killing at the cost of throughput (increased spool IO). Setting this option to
+`"ShrinkData"` or `"NoShrink"` will allow the system to use more memory and
+reduce pressure on the spool, but you increase the risk of memory usage
+exceeding limits and being targeted by the OOM killer if there is a burst
+in your workload.
+
+See also:
+* [no_memory_reduction_policy](no_memory_reduction_policy.md)
+* [Memory Management](../../memory.md)

docs/reference/kumo/make_egress_path/no_memory_reduction_policy.md

@@ -0,0 +1,38 @@
+---
+tags:
+ - memory
+---
+
+# no_memory_reduction_policy
+
+{{since('dev')}}
+
+Specifies what action should be taken when a message is added to the ready
+queue corresponding to this egress path when the memory usage is above the *soft
+memory limit*.  When the system is in this state, additional active measures
+will also be applied to reduce overall memory consumption.
+
+Possible values for this option are:
+
+* `"ShrinkDataAndMeta"` - this is the default (and is the implicit action for
+  older versions of KumoMTA).  Both the message data and metadata will be saved
+  (if modified since the prior save, or if the message has not yet been saved
+  to spool), then both will be released, freeing up the corresponding memory.
+* `"ShrinkData"` - Both the message data and metadata will be saved
+  (if modified since the prior save, or if the message has not yet been saved
+  to spool), then just the message data will be released, freeing up that memory.
+  The metadata will be preserved.
+* `"NoShrink"` - do not save or free up any message memory.
+
+This setting allows you more control in the trade-off of memory usage against
+spool IO pressure. The default is relatively conservative, aiming to avoid OOM
+killing at the cost of throughput (increased spool IO). Setting this option to
+`"ShrinkData"` or `"NoShrink"` will allow the system to use more memory and
+reduce pressure on the spool, but you increase the risk of memory usage
+exceeding limits and being targeted by the OOM killer if there is a burst
+in your workload.
+
+See also:
+* [low_memory_reduction_policy](low_memory_reduction_policy.md)
+* [Memory Management](../../memory.md)
+