add provisional knob to control how aggressive we are at open connections

I'm not sure if I want to keep this, but it is currently useful for
testing purposes.

The previous logic was to skip opening an SMTP connection in a
newly constructed Dispatcher if no messages were immediately
available in the ready queue.

This logic makes a lot of sense in a multi-node deployment
with shared throttles, so that other nodes have an opportunity
to open connections to drain their own queues, but when testing
a single node it leaves some throughput on the table.

This boolean option allows selecting whether we aggressively
continue to open an SMTP connection even if we may not have
a message ready to send yet, or, if left at the default `false`
value, continue with the original more conservative logic.

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

@@ -98,6 +98,10 @@ pub struct EgressPathConfig {
 
     #[serde(default)]
     pub suspended: bool,
+
+    // TODO: decide if we want to keep this and then document
+    #[serde(default)]
+    pub aggressive_connection_opening: bool,
 }
 
 #[cfg(feature = "lua")]
@@ -125,6 +129,7 @@ impl Default for EgressPathConfig {
             smtp_auth_plain_username: None,
             smtp_auth_plain_password: None,
             suspended: false,
+            aggressive_connection_opening: false,
         }
     }
 }

crates/kumod/src/ready_queue.rs

@@ -628,12 +628,20 @@ impl Dispatcher {
             }
         };
 
-        dispatcher.obtain_message().await;
-        if dispatcher.msg.is_none() {
-            // We raced with another dispatcher and there is no
-            // more work to be done; no need to open a new connection.
-            dispatcher.lease.release().await;
-            return Ok(());
+        // We get better throughput by being more aggressive with establishing
+        // connections.
+        if !dispatcher
+            .path_config
+            .borrow()
+            .aggressive_connection_opening
+        {
+            dispatcher.obtain_message().await;
+            if dispatcher.msg.is_none() {
+                // We raced with another dispatcher and there is no
+                // more work to be done; no need to open a new connection.
+                dispatcher.lease.release().await;
+                return Ok(());
+            }
         }
 
         let mut connection_failures = vec![];