diff --git a/crates/collab-net/src/manage.rs b/crates/collab-net/src/manage.rs index 04c718c5d..aa73a2ec5 100644 --- a/crates/collab-net/src/manage.rs +++ b/crates/collab-net/src/manage.rs @@ -206,7 +206,17 @@ 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. + // 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 { + 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}")), @@ -251,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() {