address review: use the request-reply timeout for the reply deadline

crates/collab-net/src/manage.rs

@@ -209,7 +209,9 @@ fn reply<T>(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() {