fix(http): don't back off from healthy SSE reconnects, and don't log them as errors

crates/turbomcp-http/src/lib.rs

@@ -92,7 +92,10 @@
 mod transport;
 
 // Re-export the transport implementation
-pub use transport::{RetryPolicy, StreamableHttpClientConfig, StreamableHttpClientTransport};
+pub use transport::{
+    DEFAULT_SSE_HEALTHY_STREAM_THRESHOLD, RetryPolicy, StreamableHttpClientConfig,
+    StreamableHttpClientTransport,
+};
 
 // Re-export common types from traits crate for convenience
 pub use turbomcp_transport_traits::{

crates/turbomcp-http/src/transport.rs

@@ -18,7 +18,7 @@ use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::sync::atomic::Ordering;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use tokio::sync::{Mutex, RwLock, mpsc};
 use tracing::{debug, error, info, warn};
 
@@ -83,6 +83,49 @@ impl Default for RetryPolicy {
     }
 }
 
+/// How long a standalone SSE stream must stay up before it counts as a *successful* connection
+/// rather than a failed one.
+///
+/// Connecting is not the same as working. A server that accepts the GET and then immediately closes
+/// the stream — no keepalive, or it does not really support the standalone channel — used to reset
+/// the backoff counter on every accept, so the client reconnected with **zero** delay, forever. In
+/// the field that produced ~50 reconnects a minute on a completely idle connection (93.5k in 24h),
+/// which survives functionally but buries every other diagnostic in the log.
+///
+/// Streams that do real work run far longer than this, so a genuine reconnect after a deploy still
+/// gets a full set of fresh attempts.
+/// Default for [`StreamableHttpClientConfig::sse_healthy_stream_threshold`].
+///
+/// Well below a typical server idle timeout on purpose. Servers commonly close an idle SSE stream
+/// on a round number — 30s and 60s are both common — and a threshold at or above that would
+/// classify every ordinary cycle as a failure, accumulate backoff against normal operation, and
+/// eventually abandon the channel. Measured against one such server: streams ended at 29.9999s,
+/// every time.
+pub const DEFAULT_SSE_HEALTHY_STREAM_THRESHOLD: Duration = Duration::from_secs(10);
+
+/// Whether a stream that stayed up for `uptime` did real work.
+///
+/// The single judgement this file makes about a closed stream, used twice: to decide whether the
+/// backoff counter resets, and to decide whether the reconnect is worth logging loudly. A server
+/// that closes idle streams on a timer is behaving normally and should be neither backed off from
+/// nor reported as an error.
+fn stream_was_healthy(uptime: Duration, threshold: Duration) -> bool {
+    uptime >= threshold
+}
+
+/// The attempt counter after a stream ends, given how long it was up.
+///
+/// Separated from the loop so the rule is stated once and can be tested without a server: a stream
+/// that lasted is a success and clears the backoff; one that collapsed immediately is a failure and
+/// must count as one, or backoff never engages.
+fn next_attempt_after_stream_end(previous: u32, uptime: Duration, threshold: Duration) -> u32 {
+    if stream_was_healthy(uptime, threshold) {
+        0
+    } else {
+        previous.saturating_add(1)
+    }
+}
+
 impl RetryPolicy {
     pub(crate) fn delay(&self, attempt: u32) -> Option<Duration> {
         match self {
@@ -178,6 +221,16 @@ pub struct StreamableHttpClientConfig {
     /// the SSE task breaks and the reconnect loop takes over. Set generously —
     /// the SSE protocol tolerates long idle periods between events. Default: 5 minutes.
     pub sse_read_timeout: Duration,
+
+    /// How long a standalone SSE stream must stay up to count as having done real work.
+    ///
+    /// Below this, a stream that ends is treated as a failed attempt: backoff accrues and the
+    /// reconnect is logged as a warning. At or above it, the stream is considered to have worked —
+    /// the counter resets and the reconnect is routine (`debug`).
+    ///
+    /// Keep it comfortably under the server's idle timeout. See
+    /// [`DEFAULT_SSE_HEALTHY_STREAM_THRESHOLD`].
+    pub sse_healthy_stream_threshold: Duration,
 }
 
 impl Default for StreamableHttpClientConfig {
@@ -194,6 +247,7 @@ impl Default for StreamableHttpClientConfig {
             limits: LimitsConfig::default(),
             tls: TlsConfig::default(),
             sse_read_timeout: Duration::from_secs(300),
+            sse_healthy_stream_threshold: DEFAULT_SSE_HEALTHY_STREAM_THRESHOLD,
         }
     }
 }
@@ -635,7 +689,9 @@ impl StreamableHttpClientTransport {
 
                     info!("SSE connection established");
                     *state.write().await = TransportState::Connected;
-                    attempt = 0; // Reset attempt counter on success
+                    // Deliberately *not* resetting `attempt` here: accepting the GET is not
+                    // evidence the stream works. That is decided below, from how long it lasted.
+                    let connected_at = Instant::now();
 
                     // Process SSE stream
                     let mut stream = response.bytes_stream();
@@ -705,13 +761,41 @@ impl StreamableHttpClientTransport {
                                 }
                             }
                             Err(e) => {
-                                error!("Error reading SSE stream: {}", e);
+                                // Not necessarily a fault: a server closing an idle stream
+                                // surfaces here as a decode error. Whether that mattered is
+                                // decided below, from how long the stream lasted — logging it as
+                                // an error unconditionally reports normal operation as a failure.
+                                if stream_was_healthy(
+                                    connected_at.elapsed(),
+                                    config.sse_healthy_stream_threshold,
+                                ) {
+                                    debug!("SSE stream closed by server: {}", e);
+                                } else {
+                                    error!("Error reading SSE stream: {}", e);
+                                }
                                 break;
                             }
                         }
                     }
 
-                    warn!("SSE stream ended");
+                    let uptime = connected_at.elapsed();
+                    let healthy = stream_was_healthy(uptime, config.sse_healthy_stream_threshold);
+                    attempt = next_attempt_after_stream_end(
+                        attempt,
+                        uptime,
+                        config.sse_healthy_stream_threshold,
+                    );
+                    if healthy {
+                        // A server that closes idle streams on a timer is behaving normally, and
+                        // this is the client doing its job. Reporting it at warn/error once per
+                        // cycle per peer is what buried real diagnostics under log rotation.
+                        debug!("SSE stream ended after {:?}; reconnecting", uptime);
+                    } else {
+                        warn!(
+                            "SSE stream ended after only {:?} (attempt {}); backing off",
+                            uptime, attempt
+                        );
+                    }
                     *state.write().await = TransportState::Disconnected;
                 }
                 Err(e) => {
@@ -1242,6 +1326,10 @@ impl Transport for StreamableHttpClientTransport {
 
 #[cfg(test)]
 mod tests {
+    /// Threshold used by the rule tests; the production default is
+    /// [`DEFAULT_SSE_HEALTHY_STREAM_THRESHOLD`].
+    const T: Duration = DEFAULT_SSE_HEALTHY_STREAM_THRESHOLD;
+
     use super::*;
 
     #[test]
@@ -1281,6 +1369,235 @@ mod tests {
         assert_eq!(policy.delay(3), None);
     }
 
+    /// Drives the **real** reconnect loop against a server that accepts the GET and immediately
+    /// closes the stream — the exact shape that produced ~50 reconnects/minute in the field.
+    ///
+    /// The unit tests below pin the arithmetic of `next_attempt_after_stream_end`. This one exists
+    /// because that is not the same claim: it proves the loop *as written* stops hammering. Run
+    /// against the pre-fix code (reset `attempt` on connect instead of on uptime) it counts
+    /// connections in the hundreds and fails.
+    ///
+    /// A raw `TcpListener` rather than a server framework, so the test adds no dependency and
+    /// models "accept, send headers, hang up" precisely.
+    #[tokio::test]
+    async fn a_server_that_closes_the_stream_immediately_does_not_get_hammered() {
+        use std::sync::Arc as StdArc;
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        use tokio::io::AsyncWriteExt;
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let connections = StdArc::new(AtomicUsize::new(0));
+
+        let accepted = StdArc::clone(&connections);
+        let server = tokio::spawn(async move {
+            loop {
+                let Ok((mut sock, _)) = listener.accept().await else {
+                    break;
+                };
+                accepted.fetch_add(1, Ordering::Relaxed);
+                // Handled off the accept loop so the listener keeps up; otherwise every other
+                // connection is refused and the client takes the connect-error branch instead.
+                tokio::spawn(async move {
+                    let _ = sock
+                        .write_all(
+                            b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
+                        )
+                        .await;
+                    let _ = sock.flush().await;
+                    // Let the client receive the complete response before hanging up. Dropping
+                    // immediately makes reqwest report a *connect* error, which takes the `Err`
+                    // branch that already backs off — so the success path under test never runs.
+                    tokio::time::sleep(Duration::from_millis(50)).await;
+                    drop(sock);
+                });
+            }
+        });
+
+        let config = StreamableHttpClientConfig {
+            base_url: format!("http://{addr}"),
+            retry_policy: RetryPolicy::Exponential {
+                base: Duration::from_millis(200),
+                max_delay: Duration::from_secs(5),
+                max_attempts: None, // never give up, so this measures rate and not exhaustion
+            },
+            ..Default::default()
+        };
+        let (tx, _rx) = mpsc::channel(16);
+        let task = tokio::spawn(StreamableHttpClientTransport::sse_connection_task(
+            format!("http://{addr}/mcp"),
+            config,
+            HttpClient::new(),
+            Arc::new(RwLock::new(TransportState::Disconnected)),
+            tx,
+            Arc::new(RwLock::new(None)),
+            Arc::new(RwLock::new(None)),
+            Arc::new(RwLock::new(None)),
+        ));
+
+        tokio::time::sleep(Duration::from_secs(2)).await;
+        task.abort();
+        server.abort();
+
+        let count = connections.load(Ordering::Relaxed);
+        // Measured: this loop does ~3-5 reconnects in two seconds with backoff engaged, and
+        // **24,118** with the pre-fix behaviour restored (reset the counter on connect rather
+        // than on uptime). The bound sits far above the former and far below the latter, so it
+        // discriminates without being flaky on a slow machine.
+        assert!(
+            count <= 50,
+            "reconnects must be rate-limited by backoff; saw {count} in 2s              (pre-fix behaviour produces ~24k)"
+        );
+        assert!(
+            count >= 1,
+            "the client should still have tried to connect at least once"
+        );
+    }
+
+    /// A server that holds the stream open past the threshold and then closes it is behaving
+    /// normally, and must NOT accrue backoff.
+    ///
+    /// This is the case measured in production: streams ended at 29.9999s, every time, against a
+    /// server whose idle timeout is 30s. A threshold at or above that classified every ordinary
+    /// cycle as a failure — backoff accumulated against normal operation and the client was on
+    /// course to hit `max_attempts` and abandon the channel altogether.
+    ///
+    /// Asserted through the real loop rather than the rule alone, because the rule was already
+    /// correct in isolation and the bug was in the value it was given.
+    #[tokio::test]
+    async fn a_stream_that_lives_past_the_threshold_is_not_treated_as_a_failure() {
+        use std::sync::Arc as StdArc;
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        use tokio::io::AsyncWriteExt;
+
+        // Server holds each stream open for 300ms, then closes — "long-lived" relative to the
+        // 100ms threshold below, exactly as 30s is to a 10s default.
+        const HOLD: Duration = Duration::from_millis(300);
+        const THRESHOLD: Duration = Duration::from_millis(100);
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let connections = StdArc::new(AtomicUsize::new(0));
+
+        let accepted = StdArc::clone(&connections);
+        let server = tokio::spawn(async move {
+            loop {
+                let Ok((mut sock, _)) = listener.accept().await else {
+                    break;
+                };
+                accepted.fetch_add(1, Ordering::Relaxed);
+                tokio::spawn(async move {
+                    let _ = sock
+                        .write_all(
+                            b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\nTransfer-Encoding: chunked\r\n\r\n",
+                        )
+                        .await;
+                    let _ = sock.flush().await;
+                    tokio::time::sleep(HOLD).await;
+                    drop(sock);
+                });
+            }
+        });
+
+        let config = StreamableHttpClientConfig {
+            base_url: format!("http://{addr}"),
+            sse_healthy_stream_threshold: THRESHOLD,
+            retry_policy: RetryPolicy::Exponential {
+                base: Duration::from_secs(5), // huge, so any backoff at all is unmistakable
+                max_delay: Duration::from_secs(30),
+                max_attempts: None,
+            },
+            ..Default::default()
+        };
+        let (tx, _rx) = mpsc::channel(16);
+        let task = tokio::spawn(StreamableHttpClientTransport::sse_connection_task(
+            format!("http://{addr}/mcp"),
+            config,
+            HttpClient::new(),
+            Arc::new(RwLock::new(TransportState::Disconnected)),
+            tx,
+            Arc::new(RwLock::new(None)),
+            Arc::new(RwLock::new(None)),
+            Arc::new(RwLock::new(None)),
+        ));
+
+        tokio::time::sleep(Duration::from_millis(1500)).await;
+        task.abort();
+        server.abort();
+
+        // Each cycle costs ~300ms, so ~4 fit in 1.5s. If the threshold mis-classified these as
+        // failures the 5s backoff would engage and only the first connection would ever happen.
+        let count = connections.load(Ordering::Relaxed);
+        assert!(
+            count >= 3,
+            "a healthy stream that ends must reconnect promptly, not back off; saw {count} in 1.5s"
+        );
+    }
+
+    /// A stream that collapses immediately must count as a failed attempt.
+    ///
+    /// This is the whole bug: the old code cleared the counter the moment the GET was accepted, so
+    /// a server that accepted and instantly closed produced an unthrottled reconnect loop — the
+    /// `if attempt > 0` guard on the sleep meant zero delay, every time, forever.
+    #[test]
+    fn a_stream_that_ends_immediately_counts_as_a_failed_attempt() {
+        assert_eq!(next_attempt_after_stream_end(0, Duration::ZERO, T), 1);
+        assert_eq!(
+            next_attempt_after_stream_end(3, Duration::from_millis(50), T),
+            4
+        );
+        // Just under the bar is still a failure — no "close enough".
+        assert_eq!(
+            next_attempt_after_stream_end(1, T - Duration::from_millis(1), T),
+            2
+        );
+    }
+
+    /// A stream that did real work clears the backoff, so a later reconnect (a deploy, say) starts
+    /// from a full set of attempts rather than inheriting an old count.
+    #[test]
+    fn a_long_lived_stream_resets_the_backoff() {
+        assert_eq!(
+            next_attempt_after_stream_end(7, T, T),
+            0,
+            "the threshold itself must count as healthy"
+        );
+        assert_eq!(
+            next_attempt_after_stream_end(9, Duration::from_secs(3600), T),
+            0
+        );
+    }
+
+    /// The counter feeds `delay()`, whose `max_attempts` eventually gives up. Incrementing must not
+    /// wrap round to 0 and restart the storm it was added to stop.
+    #[test]
+    fn the_attempt_counter_saturates_rather_than_wrapping() {
+        assert_eq!(
+            next_attempt_after_stream_end(u32::MAX, Duration::ZERO, T),
+            u32::MAX
+        );
+    }
+
+    /// Ties the rule back to the observable it exists to fix: with the counter rising, the policy
+    /// hands back real delays instead of the zero-delay spin.
+    #[test]
+    fn a_flapping_stream_actually_earns_a_delay() {
+        let policy = RetryPolicy::Exponential {
+            base: Duration::from_secs(1),
+            max_delay: Duration::from_secs(60),
+            max_attempts: Some(10),
+        };
+        let mut attempt = 0u32;
+        for _ in 0..3 {
+            attempt = next_attempt_after_stream_end(attempt, Duration::from_millis(10), T);
+        }
+        assert_eq!(attempt, 3);
+        assert!(
+            policy.delay(attempt).unwrap() >= Duration::from_secs(3),
+            "three straight collapses must buy real backoff, not another immediate retry"
+        );
+    }
+
     #[test]
     fn test_retry_policy_exponential() {
         let policy = RetryPolicy::Exponential {