From e29ebb7c0993289506a35bcd881c206c2ccad078 Mon Sep 17 00:00:00 2001 From: yuwk <1729065730@qq.com> Date: Mon, 31 Aug 2026 17:23:09 +0800 Subject: [PATCH 1/2] fix(sse): saturate exponential reconnect backoff ExponentialBackoff::retry computed the reconnect multiplier with 2u32.pow(current_times). With max_times unset, current_times can reach the bit width, panicking in debug builds and wrapping to a zero delay in release builds for long-lived SSE clients. Use saturating_pow and Duration::saturating_mul so the delay stays monotonic and panic-free. --- .../src/transport/common/client_side_sse.rs | 41 ++++++++++++++++++- 1 file changed, 40 insertions(+), 1 deletion(-) diff --git a/crates/rmcp/src/transport/common/client_side_sse.rs b/crates/rmcp/src/transport/common/client_side_sse.rs index e668d63df..e98ea8f6e 100644 --- a/crates/rmcp/src/transport/common/client_side_sse.rs +++ b/crates/rmcp/src/transport/common/client_side_sse.rs @@ -227,7 +227,12 @@ impl SseRetryPolicy for ExponentialBackoff { { return None; } - Some(self.base_duration * (2u32.pow(current_times as u32))) + // `current_times` is unbounded when `max_times` is unset, so the exponent can reach + // the bit width. Saturate the multiplier at `u32::MAX` and use saturating multiplication + // for the base duration so a long-lived SSE client gets a monotonic, panic-free delay + // instead of an overflow panic (debug) or a wrapped-to-zero backoff (release). + let multiplier = 2u32.saturating_pow(current_times as u32); + Some(self.base_duration.saturating_mul(multiplier)) } } @@ -775,4 +780,38 @@ mod tests { assert!(stream.next().await.is_none()); assert_eq!(attempts.load(Ordering::Relaxed), 0); } + + #[test] + fn exponential_backoff_saturates_at_high_retry_counts() { + // With `max_times` unset, `current_times` can reach the bit width. The old + // `2u32.pow(current_times)` panicked in debug builds and wrapped in release; + // the saturating implementation must return a monotonic, non-zero delay instead. + let policy = ExponentialBackoff::default(); + let mut previous = Duration::ZERO; + for current_times in [31usize, 32, 63, 64, 100] { + let delay = policy + .retry(current_times) + .expect("unbounded policy never gives up"); + assert!( + !delay.is_zero(), + "delay must stay non-zero at {current_times}" + ); + assert!( + delay >= previous, + "delay must stay monotonic at {current_times}" + ); + previous = delay; + } + } + + #[test] + fn exponential_backoff_respects_max_times() { + let policy = ExponentialBackoff { + max_times: Some(3), + base_duration: Duration::from_millis(1), + }; + assert!(policy.retry(0).is_some()); + assert!(policy.retry(2).is_some()); + assert!(policy.retry(3).is_none()); + } } From 80cbaeba84403925b038665c3c427d0762e80c2a Mon Sep 17 00:00:00 2001 From: yuwk <1729065730@qq.com> Date: Tue, 1 Sep 2026 11:04:29 +0800 Subject: [PATCH 2/2] fix(sse): cap exponential reconnect backoff at a bounded max delay Saturating the multiplier alone can still yield decades-long sleeps once current_times reaches the bit width, pinning the stream in tokio::time::sleep without reconnecting or terminating. Add an optional max_delay (default 30s) that clamps the computed delay, keeping the backoff monotonic and panic-free while guaranteeing the client retries. --- .../src/transport/common/client_side_sse.rs | 56 +++++++++++++++++-- .../src/transport/streamable_http_client.rs | 2 + 2 files changed, 54 insertions(+), 4 deletions(-) diff --git a/crates/rmcp/src/transport/common/client_side_sse.rs b/crates/rmcp/src/transport/common/client_side_sse.rs index e98ea8f6e..ccf332233 100644 --- a/crates/rmcp/src/transport/common/client_side_sse.rs +++ b/crates/rmcp/src/transport/common/client_side_sse.rs @@ -205,10 +205,16 @@ impl Default for FixedInterval { pub struct ExponentialBackoff { pub max_times: Option, pub base_duration: Duration, + /// Upper bound on a single reconnect delay. The unbounded doubling policy can otherwise + /// produce delays of decades (once the multiplier saturates), which would pin the stream in + /// `tokio::time::sleep` forever — neither reconnecting nor terminating. Capping keeps the + /// backoff monotonic and panic-free while guaranteeing the client actually retries. + pub max_delay: Option, } impl ExponentialBackoff { pub const DEFAULT_DURATION: Duration = Duration::from_millis(1000); + pub const DEFAULT_MAX_DELAY: Duration = Duration::from_secs(30); } impl Default for ExponentialBackoff { @@ -216,6 +222,7 @@ impl Default for ExponentialBackoff { Self { max_times: None, base_duration: Self::DEFAULT_DURATION, + max_delay: Some(Self::DEFAULT_MAX_DELAY), } } } @@ -229,10 +236,14 @@ impl SseRetryPolicy for ExponentialBackoff { } // `current_times` is unbounded when `max_times` is unset, so the exponent can reach // the bit width. Saturate the multiplier at `u32::MAX` and use saturating multiplication - // for the base duration so a long-lived SSE client gets a monotonic, panic-free delay - // instead of an overflow panic (debug) or a wrapped-to-zero backoff (release). + // for the base duration so the delay stays monotonic and panic-free instead of an + // overflow panic (debug) or a wrapped-to-zero backoff (release). let multiplier = 2u32.saturating_pow(current_times as u32); - Some(self.base_duration.saturating_mul(multiplier)) + let delay = self.base_duration.saturating_mul(multiplier); + Some(match self.max_delay { + Some(max_delay) => delay.min(max_delay), + None => delay, + }) } } @@ -786,7 +797,11 @@ mod tests { // With `max_times` unset, `current_times` can reach the bit width. The old // `2u32.pow(current_times)` panicked in debug builds and wrapped in release; // the saturating implementation must return a monotonic, non-zero delay instead. - let policy = ExponentialBackoff::default(); + let policy = ExponentialBackoff { + max_times: None, + base_duration: Duration::from_millis(1), + max_delay: None, + }; let mut previous = Duration::ZERO; for current_times in [31usize, 32, 63, 64, 100] { let delay = policy @@ -804,11 +819,44 @@ mod tests { } } + #[test] + fn exponential_backoff_caps_delay_at_max_delay() { + // The default cap keeps the unbounded doubling policy from producing decades-long + // sleeps once the multiplier saturates. The delay must grow monotonically, stop at + // the configured ceiling, and never exceed it. + let policy = ExponentialBackoff { + max_times: None, + base_duration: Duration::from_secs(1), + max_delay: Some(Duration::from_secs(30)), + }; + let mut previous = Duration::ZERO; + for current_times in [0usize, 1, 2, 3, 4, 5, 10, 32, 64, 100] { + let delay = policy + .retry(current_times) + .expect("unbounded policy never gives up"); + assert!( + delay >= previous, + "delay must stay monotonic at {current_times}" + ); + assert!( + delay <= Duration::from_secs(30), + "delay must respect max_delay at {current_times}" + ); + previous = delay; + } + // Beyond the ceiling the delay stays pinned at max_delay. + assert_eq!( + policy.retry(100).expect("never gives up"), + Duration::from_secs(30) + ); + } + #[test] fn exponential_backoff_respects_max_times() { let policy = ExponentialBackoff { max_times: Some(3), base_duration: Duration::from_millis(1), + max_delay: None, }; assert!(policy.retry(0).is_some()); assert!(policy.retry(2).is_some()); diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index d702bc1ca..6f27abf90 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -2293,6 +2293,7 @@ mod tests { Arc::new(ExponentialBackoff { max_times: Some(1), base_duration: Duration::ZERO, + max_delay: None, }), ); let mut stream = std::pin::pin!(stream); @@ -2397,6 +2398,7 @@ mod tests { Arc::new(ExponentialBackoff { max_times: Some(1), base_duration: Duration::ZERO, + max_delay: None, }), );