refactor(dispatch): per-mode consumer idle timeout (10s for per-message)

Per-message mode (cap=1) doesn't benefit from holding consumers across
message gaps — there is no batch window to preserve — so a 5-minute
idle timeout left consumer tasks lingering long after they were useful.
Add PER_MESSAGE_CONSUMER_IDLE_TIMEOUT (10s), wire it through main.rs
based on each adapter's message_processing_mode, and drop the unused
Dispatcher::new wrapper.

By Little's Law (steady-state idle count = arrival rate × idle window),
this cuts per-message-mode idle dispatcher footprint by 30x for the
same arrival rate while keeping batched modes' 5-minute window so
between-trigger lanes aren't torn down on every message.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: brettchien

src/dispatch.rs

@@ -169,12 +169,22 @@ impl DispatchTarget for AdapterRouter {
 // Dispatcher
 // ---------------------------------------------------------------------------
 
-/// Default idle timeout for per-thread consumer tasks. When no message arrives
-/// within this window the consumer exits, allowing `per_thread` map cleanup on
-/// the next `submit` (via `SendError` → `try_evict_locked`). Prevents unbounded
-/// task/memory growth from one-shot thread keys (e.g. Slack non-thread messages).
+/// Default idle timeout for per-thread consumer tasks in batched modes (Thread / Lane).
+/// When no message arrives within this window the consumer exits, allowing `per_thread`
+/// map cleanup on the next `submit` (via `SendError` → `try_evict_locked`). Prevents
+/// unbounded task/memory growth from one-shot thread keys (e.g. Slack non-thread messages).
+///
+/// Batched modes need a longer window so a lane that's between trigger arrivals isn't
+/// torn down and respawned on every message.
 pub const DEFAULT_CONSUMER_IDLE_TIMEOUT: Duration = Duration::from_secs(300);
 
+/// Idle timeout for per-message mode (cap=1, no batching). Per-message dispatchers
+/// don't benefit from holding consumers across message gaps — there is no batch
+/// window to preserve — so a much shorter timeout reduces idle resource footprint
+/// from one-shot thread keys (Little's Law: steady-state idle count = arrival rate
+/// × idle window).
+pub const PER_MESSAGE_CONSUMER_IDLE_TIMEOUT: Duration = Duration::from_secs(10);
+
 /// Per-thread message dispatcher for batched mode.
 ///
 /// Constructed once in `main.rs` and shared via `Arc`. Platform adapters call
@@ -194,23 +204,9 @@ pub struct Dispatcher {
 }
 
 impl Dispatcher {
-    pub fn new(
-        target: Arc<dyn DispatchTarget>,
-        max_buffered_messages: usize,
-        max_batch_tokens: usize,
-        grouping: BatchGrouping,
-    ) -> Self {
-        Self::with_idle_timeout(
-            target,
-            max_buffered_messages,
-            max_batch_tokens,
-            grouping,
-            DEFAULT_CONSUMER_IDLE_TIMEOUT,
-        )
-    }
-
-    /// Like `new`, but with a custom consumer idle timeout. Test-only knob —
-    /// production code should use `new` (which applies `DEFAULT_CONSUMER_IDLE_TIMEOUT`).
+    /// Construct a dispatcher with an explicit consumer idle timeout. Per-mode
+    /// callers in `main.rs` pass `PER_MESSAGE_CONSUMER_IDLE_TIMEOUT` for cap=1
+    /// dispatchers and `DEFAULT_CONSUMER_IDLE_TIMEOUT` for batched modes.
     pub fn with_idle_timeout(
         target: Arc<dyn DispatchTarget>,
         max_buffered_messages: usize,
@@ -1027,7 +1023,7 @@ mod tests {
             crate::config::ReactionsConfig::default(),
             crate::markdown::TableMode::Off,
         ));
-        Dispatcher::new(router, 10, 24_000, grouping)
+        Dispatcher::with_idle_timeout(router, 10, 24_000, grouping, DEFAULT_CONSUMER_IDLE_TIMEOUT)
     }
 
     #[tokio::test]
@@ -1398,7 +1394,7 @@ mod tests {
         // whose consumer is still parked but whose rx has been dropped.
         let mock = Arc::new(MockDispatchTarget::new());
         let target: Arc<dyn DispatchTarget> = mock.clone();
-        let d = Dispatcher::new(target, 10, 24_000, BatchGrouping::Thread);
+        let d = Dispatcher::with_idle_timeout(target, 10, 24_000, BatchGrouping::Thread, DEFAULT_CONSUMER_IDLE_TIMEOUT);
         let adapter: Arc<dyn ChatAdapter> = Arc::new(MockChatAdapter);
 
         let key = "mock:T".to_string();

src/main.rs

@@ -202,19 +202,20 @@ async fn main() -> anyhow::Result<()> {
         // Dispatcher is the sole serialization path for all modes. Message = cap 1
         // (each message dispatches alone, FIFO). Thread / Lane = configured cap;
         // grouping decides whether senders share a buffer or get their own lane.
-        let (slack_cap, slack_grouping) = match slack_cfg.message_processing_mode {
+        let (slack_cap, slack_grouping, slack_idle) = match slack_cfg.message_processing_mode {
             config::MessageProcessingMode::Message =>
-                (1, dispatch::BatchGrouping::Thread),
+                (1, dispatch::BatchGrouping::Thread, dispatch::PER_MESSAGE_CONSUMER_IDLE_TIMEOUT),
             config::MessageProcessingMode::Thread =>
-                (slack_cfg.max_buffered_messages, dispatch::BatchGrouping::Thread),
+                (slack_cfg.max_buffered_messages, dispatch::BatchGrouping::Thread, dispatch::DEFAULT_CONSUMER_IDLE_TIMEOUT),
             config::MessageProcessingMode::Lane =>
-                (slack_cfg.max_buffered_messages, dispatch::BatchGrouping::Lane),
+                (slack_cfg.max_buffered_messages, dispatch::BatchGrouping::Lane, dispatch::DEFAULT_CONSUMER_IDLE_TIMEOUT),
         };
-        let slack_dispatcher = Arc::new(dispatch::Dispatcher::new(
+        let slack_dispatcher = Arc::new(dispatch::Dispatcher::with_idle_timeout(
             router.clone(),
             slack_cap,
             slack_cfg.max_batch_tokens,
             slack_grouping,
+            slack_idle,
         ));
         dispatchers.lock().unwrap().push(slack_dispatcher.clone());
         Some(tokio::spawn(async move {
@@ -247,19 +248,20 @@ async fn main() -> anyhow::Result<()> {
         let router = router.clone();
         let shutdown_rx = shutdown_rx.clone();
         info!(url = %gw_cfg.url, "starting gateway adapter");
-        let (gw_cap, gw_grouping) = match gw_cfg.message_processing_mode {
+        let (gw_cap, gw_grouping, gw_idle) = match gw_cfg.message_processing_mode {
             config::MessageProcessingMode::Message =>
-                (1, dispatch::BatchGrouping::Thread),
+                (1, dispatch::BatchGrouping::Thread, dispatch::PER_MESSAGE_CONSUMER_IDLE_TIMEOUT),
             config::MessageProcessingMode::Thread =>
-                (gw_cfg.max_buffered_messages, dispatch::BatchGrouping::Thread),
+                (gw_cfg.max_buffered_messages, dispatch::BatchGrouping::Thread, dispatch::DEFAULT_CONSUMER_IDLE_TIMEOUT),
             config::MessageProcessingMode::Lane =>
-                (gw_cfg.max_buffered_messages, dispatch::BatchGrouping::Lane),
+                (gw_cfg.max_buffered_messages, dispatch::BatchGrouping::Lane, dispatch::DEFAULT_CONSUMER_IDLE_TIMEOUT),
         };
-        let gw_dispatcher = Arc::new(dispatch::Dispatcher::new(
+        let gw_dispatcher = Arc::new(dispatch::Dispatcher::with_idle_timeout(
             router.clone(),
             gw_cap,
             gw_cfg.max_batch_tokens,
             gw_grouping,
+            gw_idle,
         ));
         dispatchers.lock().unwrap().push(gw_dispatcher.clone());
         let params = gateway::GatewayParams {
@@ -346,19 +348,20 @@ async fn main() -> anyhow::Result<()> {
             "starting discord adapter"
         );
 
-        let (discord_cap, discord_grouping) = match discord_cfg.message_processing_mode {
+        let (discord_cap, discord_grouping, discord_idle) = match discord_cfg.message_processing_mode {
             config::MessageProcessingMode::Message =>
-                (1, dispatch::BatchGrouping::Thread),
+                (1, dispatch::BatchGrouping::Thread, dispatch::PER_MESSAGE_CONSUMER_IDLE_TIMEOUT),
             config::MessageProcessingMode::Thread =>
-                (discord_cfg.max_buffered_messages, dispatch::BatchGrouping::Thread),
+                (discord_cfg.max_buffered_messages, dispatch::BatchGrouping::Thread, dispatch::DEFAULT_CONSUMER_IDLE_TIMEOUT),
             config::MessageProcessingMode::Lane =>
-                (discord_cfg.max_buffered_messages, dispatch::BatchGrouping::Lane),
+                (discord_cfg.max_buffered_messages, dispatch::BatchGrouping::Lane, dispatch::DEFAULT_CONSUMER_IDLE_TIMEOUT),
         };
-        let discord_dispatcher = Arc::new(dispatch::Dispatcher::new(
+        let discord_dispatcher = Arc::new(dispatch::Dispatcher::with_idle_timeout(
             router.clone(),
             discord_cap,
             discord_cfg.max_batch_tokens,
             discord_grouping,
+            discord_idle,
         ));
         dispatchers.lock().unwrap().push(discord_dispatcher.clone());