fix: use minimal buffer for shadow streams and cap at 32

- Shadow streams only receive SSE keep-alive pings, so use capacity 1
  instead of full channel_capacity
- Cap shadow_txs at 32 to prevent unbounded growth from misbehaving
  clients, dropping the oldest shadow when the limit is reached
- Add test verifying primary works after exceeding shadow limit

Signed-off-by: Mohammod Al Amin Ashik <maa.ashik00@gmail.com>

Author: its-mash

crates/rmcp/src/transport/streamable_http_server/session/local.rs

@@ -567,8 +567,14 @@ impl LocalSessionWorker {
         &mut self,
         last_event_index: usize,
     ) -> Result<StreamableHttpMessageReceiver, SessionError> {
-        let (tx, rx) = tokio::sync::mpsc::channel(self.session_config.channel_capacity);
-        if self.common.tx.is_closed() {
+        let is_replacing_dead_primary = self.common.tx.is_closed();
+        let capacity = if is_replacing_dead_primary {
+            self.session_config.channel_capacity
+        } else {
+            1 // Shadow streams only need keep-alive pings
+        };
+        let (tx, rx) = tokio::sync::mpsc::channel(capacity);
+        if is_replacing_dead_primary {
             // Primary common channel is dead — replace it.
             tracing::debug!("Replacing dead common channel with new primary");
             self.common.tx = tx;
@@ -580,6 +586,15 @@ impl LocalSessionWorker {
             // that stays alive via SSE keep-alive but doesn't receive
             // notifications. This prevents competing EventSource connections
             // from killing each other's channels.
+            const MAX_SHADOW_STREAMS: usize = 32;
+
+            if self.shadow_txs.len() >= MAX_SHADOW_STREAMS {
+                tracing::warn!(
+                    shadow_count = self.shadow_txs.len(),
+                    "Shadow stream limit reached, dropping oldest"
+                );
+                self.shadow_txs.remove(0);
+            }
             tracing::debug!(
                 shadow_count = self.shadow_txs.len(),
                 "Common channel active, creating shadow stream"

crates/rmcp/tests/test_sse_concurrent_streams.rs

@@ -743,3 +743,42 @@ async fn dead_primary_replacement_replays_cached_events() {
 
     ct.cancel();
 }
+
+// ─── Tests: Shadow stream limits ─────────────────────────────────────────────
+
+/// Opening more than 32 shadow streams should not crash or reject — the server
+/// drops the oldest shadow to stay within the limit. Primary still works.
+#[tokio::test]
+async fn shadow_stream_limit_drops_oldest() {
+    let ct = CancellationToken::new();
+    let trigger = Arc::new(Notify::new());
+    let url = start_test_server(ct.clone(), trigger.clone()).await;
+    let client = reqwest::Client::new();
+
+    let session_id = initialize_session(&client, &url).await;
+    send_initialized_notification(&client, &url, &session_id).await;
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // First GET — primary
+    let get1 = open_standalone_get(&client, &url, &session_id).await;
+    assert_eq!(get1.status(), 200);
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    // Open 35 shadow streams (exceeds MAX_SHADOW_STREAMS=32)
+    let mut shadows = Vec::new();
+    for i in 0..35 {
+        let shadow = open_standalone_get(&client, &url, &session_id).await;
+        assert_eq!(shadow.status(), 200, "Shadow #{i} should succeed");
+        shadows.push(shadow);
+    }
+
+    // Primary should still receive notifications despite shadow churn
+    trigger.notify_one();
+
+    assert!(
+        wait_for_sse_event(get1, "tools/list_changed", Duration::from_secs(3)).await,
+        "Primary should still work after exceeding shadow limit"
+    );
+
+    ct.cancel();
+}