Merge pull request #1310 from dotCooCoo/fix/collab-net-loopback-tests

M9: the hello deadline refuses a connection even when a read times out early

crates/collab-net/src/code_tests.rs

@@ -87,7 +87,10 @@ fn a_local_share_is_joined_by_its_code_and_edits_sync() {
     let mut late = session("Cy");
     late.execute("collab.join", &json!({"code": shown})).unwrap();
     pump(&mut [&mut late], "the refusal", |s| matches!(s[0].live().status, Status::Failed { .. }));
-    assert!(late.live().status.label().contains("Couldn't join"), "{:?}", late.live().status);
+    // The port is free again: another program's host may have taken it, and that host refuses
+    // the code's secret.
+    let label = late.live().status.label();
+    assert!(label.contains("Couldn't join") || label.contains("The host refused"), "{label}");
 }
 
 #[test]

crates/collab-net/src/loopback_tests.rs

@@ -488,6 +488,21 @@ fn silent_connections_time_out() {
     host.stop();
 }
 
+#[test]
+fn connections_without_a_hello_are_told_why_they_end() {
+    // A socket's read timeout can end a moment before the hello's deadline; the connection still
+    // gets the refusal. Twelve at once, so that happens to some of them.
+    let (mut host, _hp, invite) = start_with(NetLimits { hello_timeout: Duration::from_secs(1), ..fast_limits() }, Limits::default());
+    let addr: SocketAddr = invite.server.host.parse().unwrap();
+    std::thread::scope(|s| {
+        let waits: Vec<_> = (0..12).map(|_| s.spawn(move || Raw::connect(addr).refusal())).collect();
+        for w in waits {
+            assert_eq!(w.join().unwrap().as_deref(), Some("no hello in time"));
+        }
+    });
+    host.stop();
+}
+
 #[test]
 fn guests_reconnect_after_a_network_failure() {
     let (mut host, mut hp, invite) = start();
@@ -540,7 +555,9 @@ fn quiet_guests_stay_and_silent_connections_are_dropped() {
     let (mut host, mut hp, invite) = start_with(limits, Limits::default());
     let probe = Probe::new();
     let mut cfg = quick(GuestConfig::new(invite.clone(), "Ada".into()));
-    cfg.limits = limits;
+    // The guest pings as often as the host expects. Its own idle timeout stays long: it also bounds
+    // the wait for the welcome, which a loaded machine can take longer than 600 ms to send.
+    cfg.limits.keepalive = limits.keepalive;
     let _g = Guest::start(cfg, probe.inbox.clone()).unwrap();
     let mut gp = probe;
     gp.wait("welcome", welcomed);
@@ -672,34 +689,42 @@ fn a_welcome_over_the_size_limit_says_the_file_is_too_large() {
 }
 
 /// How long a connection that never says hello stays open while `poke` sends it something every
-/// 300 ms (told whether it's the first time); `None`: still open after 6 s.
+/// 300 ms (told whether it's the first time); `None`: still open after 6 s. A thread of its own
+/// reads the socket and notes when the host ends the stream, so pongs queued ahead of the close
+/// and a slow poke loop don't delay it.
 fn pre_hello_lifetime(mut poke: impl FnMut(&mut Raw, bool) -> bool) -> Option<Duration> {
     let (mut host, _hp, invite) = start_with(NetLimits { hello_timeout: Duration::from_secs(1), ..fast_limits() }, Limits::default());
     let addr: SocketAddr = invite.server.host.parse().unwrap();
     let mut r = Raw::connect(addr);
+    let mut watch = r.0.get_ref().try_clone().unwrap();
     let t0 = Instant::now();
+    let (ended_tx, ended) = std::sync::mpsc::channel();
+    let reader = std::thread::spawn(move || {
+        let mut buf = [0u8; 1024];
+        let until = Instant::now() + WAIT;
+        loop {
+            match watch.read(&mut buf) {
+                Ok(0) => break,
+                Ok(_) => {}
+                Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) && Instant::now() < until => {}
+                Err(_) => break,
+            }
+        }
+        let _ = ended_tx.send(Instant::now());
+    });
     let (mut closed, mut first) = (None, true);
     while t0.elapsed() < Duration::from_secs(6) {
         if !poke(&mut r, std::mem::take(&mut first)) {
             closed = Some(t0.elapsed());
             break;
         }
-        std::thread::sleep(Duration::from_millis(300));
-        // Drain what the host sent back (pongs, the close); an error means it hung up.
-        r.0.get_mut().set_read_timeout(Some(Duration::from_millis(10))).unwrap();
-        match r.0.read() {
-            Ok(Message::Close(_)) | Err(tungstenite::Error::ConnectionClosed | tungstenite::Error::AlreadyClosed) => {
-                closed = Some(t0.elapsed());
-                break;
-            }
-            Err(tungstenite::Error::Io(e)) if !matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {
-                closed = Some(t0.elapsed());
-                break;
-            }
-            _ => {}
+        if let Ok(at) = ended.recv_timeout(Duration::from_millis(300)) {
+            closed = Some(at.saturating_duration_since(t0));
+            break;
         }
     }
     host.stop();
+    let _ = reader.join();
     closed
 }
 

crates/collab-net/src/net.rs

@@ -195,17 +195,25 @@ impl Duplex {
 
 impl Read for Duplex {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        if let Some(d) = self.deadline {
-            let left = d.saturating_duration_since(Instant::now());
-            if left.is_zero() {
-                return Err(io::ErrorKind::TimedOut.into());
+        let n = loop {
+            if let Some(d) = self.deadline {
+                let left = d.saturating_duration_since(Instant::now());
+                if left.is_zero() {
+                    return Err(io::ErrorKind::TimedOut.into());
+                }
+                self.read.set_read_timeout(Some(left))?;
             }
-            self.read.set_read_timeout(Some(left))?;
-        }
-        let n = match &self.tls {
-            Some(t) => tls::read(t, &mut self.read, &self.write, buf),
-            None => self.read.read(buf),
-        }?;
+            let read = match &self.tls {
+                Some(t) => tls::read(t, &mut self.read, &self.write, buf),
+                None => self.read.read(buf),
+            };
+            match read {
+                // The socket's timeout can run out a moment before the deadline (it counts whole
+                // milliseconds): wait out the rest, so a read times out only once it has passed.
+                Err(e) if self.deadline.is_some() && matches!(e.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut) => {}
+                read => break read?,
+            }
+        };
         if let Some(m) = self.meter.as_mut() {
             m.add(n)?;
         }

crates/collab-net/src/sessions_tests.rs

@@ -123,8 +123,11 @@ fn two_files_are_hosted_and_joined_independently() {
     host.execute("doc.switch", &json!({"index": 0})).unwrap();
     assert_eq!(host.live_summary()["link"], a["link"]);
     assert_eq!(host.live_summary()["document"], 0);
+    // Each joined file opens in front when its welcome arrives, and the two joins race: the file
+    // whose welcome came last is in front.
+    assert_eq!(guest.ds().unwrap().uid, guest.docs[2].uid, "the file that arrived last is in front");
+    guest.execute("doc.switch", &json!({"index": guest.docs.iter().position(|d| d.uid == gb).unwrap()})).unwrap();
     // An edit in A reaches the guest's copy of A only, while B is in front at the guest.
-    assert_eq!(guest.ds().unwrap().uid, gb, "the file joined last is in front");
     let in_a = rect(&mut host);
     host.execute("doc.switch", &json!({"index": 1})).unwrap();
     let in_b = rect(&mut host);