From 6e86e26d143f7968a15f03ec5d59188809007afe Mon Sep 17 00:00:00 2001 From: dotCooCoo Date: Sat, 24 Oct 2026 19:00:42 -0700 Subject: [PATCH] M9: the hello deadline refuses a connection even when a read times out early On Windows a socket's read timeout counts whole milliseconds, so a read bounded by the hello deadline can time out a moment before it. `next_message` then took the timeout for a closed connection: the client got no "no hello in time" refusal, and the socket closed only after the 2 s linger. `Duplex::read` now reads again until the deadline has passed. Tests: - `connections_without_a_hello_are_told_why_they_end` (new): twelve connections that never say hello each get the refusal. Failed 5 of 10 runs without the fix. - `pre_hello_lifetime` notes the end of the stream on a thread of its own instead of reading one message per 300 ms poke, where pongs queued ahead of the close delayed it. - `quiet_guests_stay_and_silent_connections_are_dropped`: the guest takes only the host's keepalive. Its 600 ms idle timeout also bounded the wait for the welcome. - `two_files_are_hosted_and_joined_independently`: the two joins race, so the file whose welcome arrived last is in front. The test checks that, then brings B to the front. - `a_local_share_is_joined_by_its_code_and_edits_sync`: once the share stops, another program's host can take the freed port and refuse the code. --- crates/collab-net/src/code_tests.rs | 5 ++- crates/collab-net/src/loopback_tests.rs | 55 ++++++++++++++++++------- crates/collab-net/src/net.rs | 28 ++++++++----- crates/collab-net/src/sessions_tests.rs | 5 ++- 4 files changed, 66 insertions(+), 27 deletions(-) diff --git a/crates/collab-net/src/code_tests.rs b/crates/collab-net/src/code_tests.rs index 7fe766ec..c2745e9a 100644 --- a/crates/collab-net/src/code_tests.rs +++ b/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] diff --git a/crates/collab-net/src/loopback_tests.rs b/crates/collab-net/src/loopback_tests.rs index d5a38d9d..fe9b4782 100644 --- a/crates/collab-net/src/loopback_tests.rs +++ b/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 { 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 } diff --git a/crates/collab-net/src/net.rs b/crates/collab-net/src/net.rs index c6628662..43e56001 100644 --- a/crates/collab-net/src/net.rs +++ b/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 { - 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)?; } diff --git a/crates/collab-net/src/sessions_tests.rs b/crates/collab-net/src/sessions_tests.rs index 27523fce..404244fa 100644 --- a/crates/collab-net/src/sessions_tests.rs +++ b/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);