From 31a60ed3606c18d8329a04e55faba6e16537419b Mon Sep 17 00:00:00 2001 From: "Dylan M. Taylor" Date: Tue, 3 Nov 2026 04:35:27 +0000 Subject: [PATCH 1/2] fix(collab-net): give a management request one deadline for its answer reply read until the server answered, and every ping, pong or control frame renewed the per-read timeout, so a library endpoint could hold a background refresh (and the shared refresh permit) open indefinitely. The whole exchange now ends at one absolute deadline, the idle timeout from when the request went out. --- crates/collab-net/src/manage.rs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/crates/collab-net/src/manage.rs b/crates/collab-net/src/manage.rs index 04c718c5d..eecbca964 100644 --- a/crates/collab-net/src/manage.rs +++ b/crates/collab-net/src/manage.rs @@ -206,7 +206,15 @@ fn send(ws: &mut Ws, msg: &ClientMsg) -> Result<(), String> { /// The next answer `pick` takes, or why there is none. fn reply(ws: &mut Ws, limits: &NetLimits, pick: impl Fn(ServerMsg) -> Option) -> Result { + // One deadline for the whole answer: pings, pongs and a message trickling in each renew the + // read timeout, so without it a server could keep a request (and the refresh permit the + // caller holds) waiting for as long as it liked. + let until = std::time::Instant::now() + limits.idle_timeout; + ws.get_mut().read_until(Some(until)); loop { + if std::time::Instant::now() >= until { + return Err("no answer from the server (it took too long)".into()); + } return match ws.read() { Ok(Message::Binary(b)) => match message::decode::(&b, limits.max_server_message) { Ok(ServerMsg::Refused { reason }) => Err(format!("{REFUSED}{reason}")), From 9aad73169324d1e2ca6dee90cb56efd5966511ca Mon Sep 17 00:00:00 2001 From: "Dylan M. Taylor" Date: Wed, 4 Nov 2026 01:29:49 +0000 Subject: [PATCH 2/2] address review: use the request-reply timeout for the reply deadline --- crates/collab-net/src/manage.rs | 25 ++++++++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/crates/collab-net/src/manage.rs b/crates/collab-net/src/manage.rs index eecbca964..aa73a2ec5 100644 --- a/crates/collab-net/src/manage.rs +++ b/crates/collab-net/src/manage.rs @@ -209,7 +209,9 @@ fn reply(ws: &mut Ws, limits: &NetLimits, pick: impl Fn(ServerMsg) -> Option< // One deadline for the whole answer: pings, pongs and a message trickling in each renew the // read timeout, so without it a server could keep a request (and the refresh permit the // caller holds) waiting for as long as it liked. - let until = std::time::Instant::now() + limits.idle_timeout; + // The request-reply bound, not the session idle one: a publish uploads a whole file and can + // legitimately take minutes to be answered (see `connect`). + let until = std::time::Instant::now() + limits.request_reply_timeout; ws.get_mut().read_until(Some(until)); loop { if std::time::Instant::now() >= until { @@ -259,6 +261,27 @@ mod tests { rx.recv_timeout(Duration::from_secs(3)).is_ok() } + /// A server that keeps pinging and never answers ends the wait at the request-reply bound. + #[test] + fn reply_gives_up_at_the_request_reply_deadline_against_a_pinging_server() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + std::thread::spawn(move || { + let (s, _) = listener.accept().unwrap(); + if let Ok(mut ws) = tungstenite::accept(s) { + while ws.send(Message::Ping(Bytes::new())).is_ok() { + std::thread::sleep(Duration::from_millis(10)); + } + } + }); + let limits = NetLimits { request_reply_timeout: Duration::from_millis(400), ..NetLimits::default() }; + let server = Endpoint { host: addr.to_string(), path: "/".into(), secure: false }; + let mut ws = connect(&server, &limits, Duration::from_secs(2)).unwrap(); + let started = std::time::Instant::now(); + assert!(reply::<()>(&mut ws, &limits, |_| None).is_err()); + assert!(started.elapsed() < Duration::from_secs(3), "{:?}", started.elapsed()); + } + /// A server that keeps pinging and never answers the close must not pin `close` forever. #[test] fn close_returns_against_a_server_that_keeps_pinging() {