From e0494962f947787e344aa8a526da77acea09c9c0 Mon Sep 17 00:00:00 2001 From: Shiloh Mangus Date: Sat, 8 Aug 2026 11:24:47 -0500 Subject: [PATCH 1/3] fix(http): back off when an SSE stream collapses instead of spinning MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The standalone SSE reconnect loop cleared its attempt counter the moment the GET was accepted: info!("SSE connection established"); attempt = 0; // Reset attempt counter on success but the sleep at the top of the loop is guarded by `if attempt > 0`. So a server that accepts the GET and then immediately closes the stream — no keepalive, or no real support for the standalone channel — produced a reconnect with **zero** delay, every time, indefinitely. Backoff only ever applied to connections that failed to *open*, never to ones that opened and collapsed. Measured on a live homelab box: ~50 reconnects a minute on a completely idle connection, 93,532 occurrences in 24 hours. Functionally survivable (tool calls go over POST and still work), but it burns the container's log budget and evicts real diagnostics under rotation — which is how it was found, while looking for something else. Connecting is not the same as working, so the counter is now decided by how long the stream lasted: 30s or more is a success and clears it, anything shorter is a failure and counts as one. Extracted as `next_attempt_after_stream_end` so the rule is stated once and testable without a server. **One behavioural change worth naming:** a stream that always collapses now exhausts `max_attempts` (default 10, ~5 minutes with the default policy) and gives up on the standalone channel, where before it retried forever. That is consistent with what the code already does for connect failures, and with the existing HTTP 405 path that logs "Continuing without standalone SSE polling" and breaks — giving up on this channel is an established, survivable outcome. A stream that ever ran normally resets the counter, so an ordinary reconnect after a deploy still gets a full set of attempts. Mutation-checked: restoring the old always-reset behaviour fails three tests, including the one that ties the rule to the observable — three straight collapses must buy a real delay rather than another immediate retry. --- crates/turbomcp-http/src/transport.rs | 107 +++++++++++++++++++++++++- 1 file changed, 104 insertions(+), 3 deletions(-) diff --git a/crates/turbomcp-http/src/transport.rs b/crates/turbomcp-http/src/transport.rs index a8c41f33..418fde9e 100644 --- a/crates/turbomcp-http/src/transport.rs +++ b/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,32 @@ 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. +const HEALTHY_STREAM_THRESHOLD: Duration = Duration::from_secs(30); + +/// 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) -> u32 { + if uptime >= HEALTHY_STREAM_THRESHOLD { + 0 + } else { + previous.saturating_add(1) + } +} + impl RetryPolicy { pub(crate) fn delay(&self, attempt: u32) -> Option { match self { @@ -635,7 +661,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(); @@ -711,7 +739,16 @@ impl StreamableHttpClientTransport { } } - warn!("SSE stream ended"); + let uptime = connected_at.elapsed(); + attempt = next_attempt_after_stream_end(attempt, uptime); + if attempt == 0 { + warn!("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) => { @@ -1281,6 +1318,70 @@ mod tests { assert_eq!(policy.delay(3), None); } + /// 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), 1); + assert_eq!( + next_attempt_after_stream_end(3, Duration::from_millis(50)), + 4 + ); + // Just under the bar is still a failure — no "close enough". + assert_eq!( + next_attempt_after_stream_end(1, HEALTHY_STREAM_THRESHOLD - Duration::from_millis(1)), + 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, HEALTHY_STREAM_THRESHOLD), + 0, + "the threshold itself must count as healthy" + ); + assert_eq!( + next_attempt_after_stream_end(9, Duration::from_secs(3600)), + 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), + 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)); + } + 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 { From e877d3849d214d72e1772afcf1c538aa620c4c3e Mon Sep 17 00:00:00 2001 From: Shiloh Mangus Date: Sat, 8 Aug 2026 11:46:35 -0500 Subject: [PATCH 2/3] test(http): prove the reconnect storm against the real loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The unit tests pinned the arithmetic of `next_attempt_after_stream_end`. That is not the same claim as "the loop stops hammering", and the difference turned out to matter: two earlier attempts at this test passed with the fix removed. Both failed for the same reason — the test server never exercised the path under test. Writing the response headers and dropping the socket immediately makes reqwest report a *connect* error, which takes the `Err` branch that already backs off correctly. The success path, where the bug lives, never ran. Instrumenting the loop showed it plainly: every iteration was `CONNECT-ERR`, never a completed response. The server now hands the connection off the accept loop, holds the socket open long enough for the client to receive a complete response, and sends `Connection: close` so reqwest does not pool a socket it is about to drop. With that, the difference is not subtle: pre-fix 24,118 connections in 2s fixed 5 connections in 2s Raw `TcpListener` rather than a server framework, so this adds no dependency. --- crates/turbomcp-http/src/transport.rs | 85 +++++++++++++++++++++++++++ 1 file changed, 85 insertions(+) diff --git a/crates/turbomcp-http/src/transport.rs b/crates/turbomcp-http/src/transport.rs index 418fde9e..1566c58b 100644 --- a/crates/turbomcp-http/src/transport.rs +++ b/crates/turbomcp-http/src/transport.rs @@ -1318,6 +1318,91 @@ 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 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 From d17c2fbd62e20ddaae40cac1222160b5501e73b2 Mon Sep 17 00:00:00 2001 From: Shiloh Mangus Date: Sat, 8 Aug 2026 12:13:53 -0500 Subject: [PATCH 3/3] fix(http): make the healthy-stream threshold configurable and quiet routine reconnects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deploying the previous commit against a real server showed the threshold was badly chosen and the remedy incomplete. **The threshold sat on the wrong side of the observed behaviour.** That server closes idle SSE streams on a 30s timer; measured uptimes were 29.9999s, every time, without variance. A 30s threshold therefore classified every ordinary cycle as a failure: backoff accrued against normal operation and the client was on course to exhaust `max_attempts` and abandon the channel — trading log noise for lost server→client notifications, which is a worse outcome than the noise. Now `sse_healthy_stream_threshold`, defaulting to 10s: comfortably under the round-number idle timeouts servers actually use, so a normal cycle resets the counter and reconnects promptly. Backoff is left guarding only the pathological case — a stream that collapses on contact, which is what the previous commit's integration test reproduces at 24k connections in 2s. **Severity now follows the same judgement.** A server closing an idle stream is behaving normally and the client reconnecting is it doing its job; reporting that at `error`/`warn` once per cycle per peer is what buried real diagnostics under log rotation. `stream_was_healthy` decides both the counter and the log level, so the two cannot disagree. A short collapse still warns. The read error is treated the same way: a server's close surfaces there as a decode error, so it is `debug` when the stream had lived and `error` only when it had not. Adds `a_stream_that_lives_past_the_threshold_is_not_treated_as_a_failure`, which asserts through the real loop that a long-lived stream reconnects promptly rather than backing off. Setting the threshold above the server's close time — the mistake this commit fixes — takes it from 4 connections in 1.5s to 1. --- crates/turbomcp-http/src/lib.rs | 5 +- crates/turbomcp-http/src/transport.rs | 159 +++++++++++++++++++++++--- 2 files changed, 149 insertions(+), 15 deletions(-) diff --git a/crates/turbomcp-http/src/lib.rs b/crates/turbomcp-http/src/lib.rs index e280c876..c532a5ef 100644 --- a/crates/turbomcp-http/src/lib.rs +++ b/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::{ diff --git a/crates/turbomcp-http/src/transport.rs b/crates/turbomcp-http/src/transport.rs index 1566c58b..f8c76213 100644 --- a/crates/turbomcp-http/src/transport.rs +++ b/crates/turbomcp-http/src/transport.rs @@ -94,15 +94,32 @@ impl Default for RetryPolicy { /// /// 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. -const HEALTHY_STREAM_THRESHOLD: Duration = Duration::from_secs(30); +/// 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) -> u32 { - if uptime >= HEALTHY_STREAM_THRESHOLD { +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) @@ -204,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 { @@ -220,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, } } } @@ -733,16 +761,35 @@ 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; } } } let uptime = connected_at.elapsed(); - attempt = next_attempt_after_stream_end(attempt, uptime); - if attempt == 0 { - warn!("SSE stream ended after {:?}; reconnecting", uptime); + 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", @@ -1279,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] @@ -1403,6 +1454,86 @@ mod tests { ); } + /// 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 @@ -1410,14 +1541,14 @@ mod tests { /// `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), 1); + assert_eq!(next_attempt_after_stream_end(0, Duration::ZERO, T), 1); assert_eq!( - next_attempt_after_stream_end(3, Duration::from_millis(50)), + 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, HEALTHY_STREAM_THRESHOLD - Duration::from_millis(1)), + next_attempt_after_stream_end(1, T - Duration::from_millis(1), T), 2 ); } @@ -1427,12 +1558,12 @@ mod tests { #[test] fn a_long_lived_stream_resets_the_backoff() { assert_eq!( - next_attempt_after_stream_end(7, HEALTHY_STREAM_THRESHOLD), + 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)), + next_attempt_after_stream_end(9, Duration::from_secs(3600), T), 0 ); } @@ -1442,7 +1573,7 @@ mod tests { #[test] fn the_attempt_counter_saturates_rather_than_wrapping() { assert_eq!( - next_attempt_after_stream_end(u32::MAX, Duration::ZERO), + next_attempt_after_stream_end(u32::MAX, Duration::ZERO, T), u32::MAX ); } @@ -1458,7 +1589,7 @@ mod tests { }; let mut attempt = 0u32; for _ in 0..3 { - attempt = next_attempt_after_stream_end(attempt, Duration::from_millis(10)); + attempt = next_attempt_after_stream_end(attempt, Duration::from_millis(10), T); } assert_eq!(attempt, 3); assert!(