Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 87 additions & 1 deletion crates/rmcp/src/transport/common/client_side_sse.rs
Original file line number Diff line number Diff line change
Expand Up @@ -205,6 +205,11 @@ impl Default for FixedInterval {
pub struct ExponentialBackoff {
pub max_times: Option<usize>,
pub base_duration: Duration,
/// Optional upper bound on a single reconnect delay. `None` (the default) preserves the
/// pre-existing unbounded doubling behavior. Once the multiplier saturates near the bit
/// width that can still produce very long sleeps, so callers that need the client to
/// actually reconnect can set `Some(...)` to clamp the delay.
pub max_delay: Option<Duration>,
}

impl ExponentialBackoff {
Expand All @@ -216,6 +221,7 @@ impl Default for ExponentialBackoff {
Self {
max_times: None,
base_duration: Self::DEFAULT_DURATION,
max_delay: None,
}
}
}
Expand All @@ -227,7 +233,16 @@ 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 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);
let delay = self.base_duration.saturating_mul(multiplier);
Some(match self.max_delay {
Some(max_delay) => delay.min(max_delay),
None => delay,
})
}
}

Expand Down Expand Up @@ -775,4 +790,75 @@ 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 {
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
.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_caps_delay_at_max_delay() {
// An explicit 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());
assert!(policy.retry(3).is_none());
}
}
2 changes: 2 additions & 0 deletions crates/rmcp/src/transport/streamable_http_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -2397,6 +2398,7 @@ mod tests {
Arc::new(ExponentialBackoff {
max_times: Some(1),
base_duration: Duration::ZERO,
max_delay: None,
}),
);

Expand Down
Loading