Skip to content

fix(moq-tokio): the retry window no longer cuts a dial short #5352

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kixelated merged 2 commits into from
Oct 13, 2026
Merged

fix(moq-tokio): the retry window no longer cuts a dial short #5352

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0f47132
fix(moq-tokio): the retry window no longer cuts a dial short
kixelated Oct 12, 2026
130a5e8
docs(moq-tokio): the retry window lets a walk under way finish
kixelated Oct 13, 2026
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
147 changes: 17 additions & 130 deletions rs/moq-tokio/src/connection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,23 +26,15 @@ const CONNECT_ATTEMPT: Duration = Duration::from_secs(5);
/// wait. Only an attempt with a later candidate to fall back on is worth
/// bounding: cutting the last one short just turns a slow connect into a failed
/// one, on every reconnect cycle.
///
/// The same goes for the retry window from [`Backoff::timeout`]: it decides
/// whether a new walk starts, not how long one may take, so a walk under way
/// tries every candidate and a handshake slower than the window still gets
/// [`crate::connect::Config::timeout`](crate::connect::Config::timeout).
fn attempt_timeout(index: usize, total: usize) -> Option<Duration> {
(index + 1 < total).then_some(CONNECT_ATTEMPT)
}

/// The retry window as a deadline for the address walk, or `None` to leave each
/// attempt to the client's own connect timeout.
///
/// Only when reconnecting. `Backoff::timeout` is how long to keep *retrying*, so
/// applying it to a one-shot dial would cut short the single attempt there will
/// ever be: a handshake slower than the retry window but well inside
/// [`crate::connect::Config::timeout`](crate::connect::Config::timeout) would fail for a
/// reason that does not apply. That deadline is the one governing a one-shot
/// dial, and it already does.
fn retry_budget(reconnect: bool, retry_start: tokio::time::Instant, timeout: Duration) -> Option<tokio::time::Instant> {
(reconnect && !timeout.is_zero()).then(|| retry_start + timeout)
}

/// Whether a failed candidate is the peer's answer for every candidate, ending the walk.
///
/// An auth rejection arrives as [`Error::Connect`] with its status already consumed, so it
Expand All @@ -54,39 +46,6 @@ fn settles_walk(err: &Error) -> bool {
.is_some_and(|status| !crate::error::status_retryable(status))
}

/// When the attempt at candidate `index` of `total` must be given up, or `None`
/// to let it run.
///
/// Two bounds, whichever comes first. [`attempt_timeout`] keeps one slow
/// candidate from eating the others' turn, and `budget` is the retry window from
/// [`Backoff::timeout`], which the walk has to respect too: bounding only the
/// non-final attempts left the last one to run to the client's connect timeout,
/// so a handful of black-holed addresses could blow through a give-up budget
/// several times over before reporting anything.
fn attempt_deadline(
index: usize,
total: usize,
now: tokio::time::Instant,
budget: Option<tokio::time::Instant>,
) -> Option<tokio::time::Instant> {
// An equal share of what's left, not all of it. Handing each candidate the
// whole remaining window lets the first couple spend it between them and
// strand a reachable address further down the list, which is the opposite of
// what walking the list is for. Recomputed per attempt, so one that fails fast
// leaves more for the rest.
let share = budget.map(|budget| {
let remaining = total - index;
now + budget.saturating_duration_since(now) / remaining as u32
});
let bound = attempt_timeout(index, total).map(|limit| now + limit);

match (bound, share) {
(Some(bound), Some(share)) => Some(bound.min(share)),
(Some(only), None) | (None, Some(only)) => Some(only),
(None, None) => None,
}
}

/// Exponential backoff configuration for reconnection attempts.
///
/// The delays carry jitter, so a fleet
Expand Down Expand Up @@ -145,6 +104,11 @@ pub struct Backoff {

/// Maximum time to spend retrying before giving up. Defaults to 10s.
///
/// No new walk of the peer's addresses starts once it has passed, but one under way finishes:
/// each candidate but the last gets up to 5s, and the last runs to
/// [`crate::connect::Config::timeout`](crate::connect::Config::timeout), so a handshake
/// slower than this still lands. Giving up can therefore take up to one whole walk longer.
///
/// Resets after a stable connection (one that outlives the initial backoff), so a flapping
/// session that reconnects then immediately drops still counts toward the timeout. Set to
/// `Some(Duration::ZERO)` for unlimited retries. `None` takes the default. When this config was
Expand Down Expand Up @@ -881,9 +845,7 @@ impl Connection {
return Err(timeout_error(timeout, retry.last_error.as_ref()));
}

let budget = retry_budget(client.reconnect, retry.start, timeout);

match Self::dial_any(shared, &client, &addrs, &mut draining, budget).await {
match Self::dial_any(shared, &client, &addrs, &mut draining).await {
Ok((addr, dialed)) => {
let url = addr.url().clone();
tracing::info!(peer = %Endpoint(&url), transport = %dialed.transport, "connected");
Expand Down Expand Up @@ -1120,20 +1082,13 @@ impl Connection {
client: &Client,
addrs: &Addrs,
draining: &mut Option<Draining>,
budget: Option<tokio::time::Instant>,
) -> crate::Result<(crate::connect::Addr, crate::client::Dialed)> {
let candidates = addrs.as_slice();
let mut last = None;
let mut transient = None;

for (index, addr) in candidates.iter().enumerate() {
let url = addr.url();
// The retry window can run out mid-walk. Stop rather than starting an
// attempt with no time to finish; the caller reports the budget error.
if budget.is_some_and(|budget| tokio::time::Instant::now() >= budget) {
break;
}

tracing::info!(peer = %Endpoint(url), "connecting");

let mut dial = std::pin::pin!(client.dial(addr.clone()));
Expand All @@ -1144,10 +1099,9 @@ impl Connection {
waiter.poll_future(dial.as_mut())
});

let deadline = attempt_deadline(index, candidates.len(), tokio::time::Instant::now(), budget);
let dialed = match deadline {
let dialed = match attempt_timeout(index, candidates.len()) {
None => dialed.await,
Some(deadline) => match tokio::time::timeout_at(deadline, dialed).await {
Some(limit) => match tokio::time::timeout(limit, dialed).await {
Ok(dialed) => dialed,
Err(_) => Err(Error::Reconnect(format!("timed out connecting to {}", Endpoint(url)))),
},
Expand All @@ -1171,12 +1125,8 @@ impl Connection {
}
}

// `Addrs` is never empty, so the only way to get here without a failure is
// the budget running out before the first attempt started.
// One unusable candidate must not turn another candidate's transient failure terminal.
Err(transient
.or(last)
.unwrap_or_else(|| Error::Reconnect("retry window elapsed before dialing".to_string())))
Err(transient.or(last).expect("Addrs is never empty"))
}

/// Poll until a session is established (the first connect, or the current one).
Expand Down Expand Up @@ -2263,7 +2213,7 @@ mod tests {

let (connected, _session) = tokio::time::timeout(
WALK_TIMEOUT,
Connection::dial_any(&shared, &client, &addrs, &mut draining, None),
Connection::dial_any(&shared, &client, &addrs, &mut draining),
)
.await
.expect("the walk must not hang")
Expand All @@ -2285,7 +2235,7 @@ mod tests {

let result = tokio::time::timeout(
WALK_TIMEOUT,
Connection::dial_any(&shared, &client, &addrs, &mut draining, None),
Connection::dial_any(&shared, &client, &addrs, &mut draining),
)
.await
.expect("the walk must not hang");
Expand Down Expand Up @@ -2313,7 +2263,7 @@ mod tests {
let shared = shared();
let mut draining = None;
// `Session` isn't `Debug`, so unwrap the failure by hand.
let Err(err) = Connection::dial_any(&shared, &client, &addrs, &mut draining, None).await else {
let Err(err) = Connection::dial_any(&shared, &client, &addrs, &mut draining).await else {
panic!("a dead address must fail");
};

Expand Down Expand Up @@ -2425,69 +2375,6 @@ mod tests {
))));
}

/// The retry window bounds the walk only when there are retries.
///
/// It is a budget for *retrying*, so spending it on the single attempt a
/// one-shot dial gets would fail a handshake that is merely slower than the
/// retry window while well inside the connect timeout that actually governs it.
#[test]
fn only_a_reconnecting_dial_spends_the_retry_window() {
let start = tokio::time::Instant::now();
let timeout = Duration::from_secs(10);

assert_eq!(retry_budget(true, start, timeout), Some(start + timeout));
assert_eq!(
retry_budget(false, start, timeout),
None,
"a one-shot dial is bounded by the connect timeout, not the retry window"
);
assert_eq!(
retry_budget(true, start, Duration::ZERO),
None,
"zero means retry forever, so nothing bounds the walk"
);
}

/// Whichever of the two bounds comes first wins, and the retry window applies
/// to every candidate including the last.
#[test]
fn the_retry_window_bounds_the_walk_too() {
let now = tokio::time::Instant::now();

// No budget: only the fixed per-candidate bound applies, and the last
// candidate runs free.
assert_eq!(attempt_deadline(0, 2, now, None), Some(now + CONNECT_ATTEMPT));
assert_eq!(attempt_deadline(1, 2, now, None), None);

// A tight budget is split, not spent by whoever asks first. Two seconds
// across two candidates is a second each, so the second one still gets a
// turn; handing the first the whole window would strand it.
let tight = now + Duration::from_secs(2);
assert_eq!(
attempt_deadline(0, 2, now, Some(tight)),
Some(now + Duration::from_secs(1))
);
// The last candidate gets what is actually left, which here is all of it
// because nothing has been spent yet in this synthetic call.
assert_eq!(attempt_deadline(1, 2, now, Some(tight)), Some(tight));

// The share is what a black-holed pair used to exhaust: with the default
// 10s window and the 5s bound, two candidates ate it and a reachable third
// was never dialed. Now each gets a third.
let default_window = now + Duration::from_secs(10);
assert_eq!(
attempt_deadline(0, 3, now, Some(default_window)),
Some(now + Duration::from_secs(10) / 3),
"a reachable third candidate must still get a turn"
);

// A budget with room to spare leaves the per-candidate bound in charge, so
// one slow address still cannot eat the others' turn.
let loose = now + Duration::from_secs(60);
assert_eq!(attempt_deadline(0, 2, now, Some(loose)), Some(now + CONNECT_ATTEMPT));
assert_eq!(attempt_deadline(1, 2, now, Some(loose)), Some(loose));
}

/// Only an attempt with somewhere to fall back on gets the fixed bound.
#[test]
fn only_a_candidate_with_a_fallback_is_bounded() {
Expand Down
65 changes: 65 additions & 0 deletions rs/moq-tokio/tests/reconnect.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,71 @@ async fn a_transient_failure_retries_until_the_budget_runs_out() {
);
}

/// The retry window decides whether a new attempt starts, not how long one may take: a handshake
/// slower than the window, but inside the connect timeout, lands on the first dial. Cutting it at
/// the window would only fail it, and every redial the same way.
#[tokio::test]
async fn a_handshake_slower_than_the_retry_window_still_lands() {
const WINDOW: Duration = Duration::from_millis(300);
const HANDSHAKE: Duration = Duration::from_secs(1);

// `tls://`, so the dial waits on the server's half of the TLS handshake.
let mut listen = moq_tokio::listen::Config::default();
listen.tcp.bind = Some("127.0.0.1:0".parse().expect("parse addr"));
listen.tcp.tls = Some(true);
listen.tls.generate = vec!["localhost".into()];
let mut server = listen
.init(Default::default())
.expect("init server")
.listen()
.await
.expect("bind tcp listener");
let server_addr = server.tcp_local_addr().expect("tcp listener bound");
let (accepted, mut sessions) = tokio::sync::mpsc::unbounded_channel();
let _server = tokio::spawn(async move {
let origin = moq_tokio::origin::spawn();
while let Some(request) = server.accept().await {
if let Ok(session) = request.with_publisher(&origin).ok().await {
let _ = accepted.send(session);
}
}
});

// Holds each connection for the handshake's length before forwarding it to the server.
let proxy = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind proxy");
let port = proxy.local_addr().expect("proxy bound").port();
let (dialed, mut dials) = tokio::sync::mpsc::unbounded_channel();
let _proxy = tokio::spawn(async move {
while let Ok((mut client, _)) = proxy.accept().await {
let _ = dialed.send(());
tokio::spawn(async move {
tokio::time::sleep(HANDSHAKE).await;
let mut upstream = tokio::net::TcpStream::connect(server_addr)
.await
.expect("connect to the server");
let _ = tokio::io::copy_bidirectional(&mut client, &mut upstream).await;
});
}
});

let mut config = moq_tokio::connect::Config::default();
config.tls.insecure = Some(true);
config.backoff.timeout = Some(WINDOW);
let client = config.init(Default::default()).expect("failed to init client");
let url: url::Url = format!("tls://127.0.0.1:{port}/").parse().expect("parse url");
let connection = client.connect(url);

tokio::select! {
session = sessions.recv() => {
session.expect("server stopped accepting");
}
res = connection.closed() => panic!("gave up mid-handshake: {res:?}"),
_ = tokio::time::sleep(Duration::from_secs(30)) => panic!("never connected"),
}
dials.recv().await.expect("proxy stopped");
assert!(dials.try_recv().is_err(), "the first dial was cut short and redialed");
}

/// The connect count advances in the same write that reports `Connected`, so a caller
/// pairing [`Connection::status`] with [`Connection::connects`] never sees a stale
/// count, however fast the redial.
Expand Down
Loading