diff --git a/crates/notebook/src/live.rs b/crates/notebook/src/live.rs index 75bddc30729038f2993d08e39cc932ddf14d5b2e..f2b4fc7714d02a253c92f156b5f6cfe8620c24f3 100644 --- a/crates/notebook/src/live.rs +++ b/crates/notebook/src/live.rs @@ -178,8 +178,7 @@ trait Pipe: Send + Sync { fn direct(&self) -> bool; /// Hears whether the peer at the other end knew the secret. fn met(&self, _met: bool) {} - /// Hears that frames arrived lost, repeated, reordered or forged, after which this end - /// meets its peers again. + /// Hears that the stream failed, after which this end meets its peers again. fn broken(&self) {} } @@ -558,20 +557,26 @@ impl Shared { let other = (met.as_ref().err()) .and_then(|error| error.get_ref()?.downcast_ref::()) .copied(); - // A peer of another version guessed nothing, the keys never being agreed. - pipe.met(met.is_ok() || other.is_some()); let (send, mut receive, hello) = match met { - Ok(met) => met, + Ok(met) => { + pipe.met(true); + met + } Err(error) => { eprintln!("Live: no meeting in {tag}: {error}"); if let Some(wire::Version(version)) = other { + // Another version guessed nothing: no keys were agreed. + pipe.met(true); self.state.lock().unwrap().outdated = Some(version); if matches!(self.room, Room::Code { owner: false, .. }) { self.stopped.store(true, Ordering::Release); } (self.events)(Event::Changed); } else if error.kind() == io::ErrorKind::InvalidData { + pipe.met(false); self.failed(); + } else { + pipe.broken(); } pipe.shutdown(); return false; @@ -651,7 +656,7 @@ impl Shared { } } }; - if let Some(error) = ended.filter(|error| error.kind() == io::ErrorKind::InvalidData) { + if let Some(error) = ended { eprintln!("Live: the connection to {name} broke ({error}); meeting again"); pipe.broken(); } @@ -698,6 +703,7 @@ impl Shared { None => Ok(()), }); if result.is_err() { + pipe.broken(); pipe.shutdown(); return; } diff --git a/crates/notebook/src/live/relay.rs b/crates/notebook/src/live/relay.rs index 74c5aa8d5f616ea21a8365fb3532fbf564d2daa2..cbc944cbd9da0bb0dad79b549f3e83c57d4687d8 100644 --- a/crates/notebook/src/live/relay.rs +++ b/crates/notebook/src/live/relay.rs @@ -551,7 +551,10 @@ impl Pipe for Relayed { } fn broken(&self) { - self.socket.hang_up(); + // A deliberate shutdown can wake the reader with EOF. + if !self.socket.links.lock().unwrap().ended.contains(&self.slot) { + self.socket.hang_up(); + } } } diff --git a/crates/notebook/src/live/tests.rs b/crates/notebook/src/live/tests.rs index 39696aa0d7ab896ba77421948f835d1039252296..1befa640fb5d0e131f19eb79099088be29f3a65b 100644 --- a/crates/notebook/src/live/tests.rs +++ b/crates/notebook/src/live/tests.rs @@ -292,6 +292,193 @@ fn peers_meet_through_a_relay() { until(&grace, <[Peer]>::is_empty); } +#[test] +fn relay_timeouts_reopen_with_fresh_keys() { + use ::relay::ws::{self, Message}; + + let (url, upstream) = relay(Default::default()); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let delayed = format!("ws://{}", listener.local_addr().unwrap()); + listener.set_nonblocking(true).unwrap(); + let (opening, openings) = mpsc::channel(); + let proxy = thread::spawn(move || { + for attempt in 0..3 { + let deadline = Instant::now() + Duration::from_secs(15); + let mut client = loop { + match listener.accept() { + Ok((client, _)) => break client, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + assert!(Instant::now() < deadline, "the relay room never reopened"); + thread::sleep(Duration::from_millis(20)); + } + Err(error) => panic!("{error}"), + } + }; + client.set_nonblocking(false).unwrap(); + client + .set_read_timeout(Some(Duration::from_secs(15))) + .unwrap(); + let mut server = TcpStream::connect(upstream).unwrap(); + let head = ws::head(&mut client).unwrap(); + server.write_all(head.as_bytes()).unwrap(); + let (mut down, mut to_client) = + (server.try_clone().unwrap(), client.try_clone().unwrap()); + let replies = thread::spawn(move || { + let _ = io::copy(&mut down, &mut to_client); + let _ = to_client.shutdown(Shutdown::Both); + }); + let mut messages = ws::Reader::new(client, 1 << 20, true); + let mut first = true; + while let Ok(message) = messages.read() { + let (opcode, data) = match message { + Message::Binary(data) => { + let slot = u32::from_be_bytes(data[..4].try_into().unwrap()); + if first && slot > 0 && slot < ::relay::GROUP { + first = false; + opening.send(data[4..].to_vec()).unwrap(); + // Keep the first opening delayed until its stream is abandoned. + if attempt == 0 { + continue; + } + } + (ws::BINARY, data) + } + Message::Text(text) => (ws::TEXT, text.into_bytes()), + Message::Ping(data) => (ws::PING, data), + Message::Pong => (ws::PONG, vec![]), + Message::Close => break, + }; + if server + .write_all(&ws::frame(opcode, &data, Some([1, 2, 3, 4]))) + .is_err() + { + break; + } + } + let _ = server.shutdown(Shutdown::Both); + replies.join().unwrap(); + } + }); + + let room = Room::Notebook([7; 16]); + let mut host = hello("Host"); + host.serves = Some([8; 16]); + let (met, host_met) = mpsc::channel(); + let (frame, frames) = mpsc::channel(); + let host = Live::start(host, &room, None, Some(&url), move |event| match event { + Event::Met(_, line) => { + let _ = met.send(line.clone()); + } + Event::Frame { kind, body, .. } => { + let _ = frame.send((kind, body.to_vec())); + } + _ => {} + }) + .unwrap(); + let (met, guest_met) = mpsc::channel(); + let (left, guest_left) = mpsc::channel(); + let (frame, guest_frames) = mpsc::channel(); + let guest = Live::start( + hello("Guest"), + &room, + None, + Some(&delayed), + move |event| match event { + Event::Met(_, line) => { + let _ = met.send(line.clone()); + } + Event::Left(_) => { + let _ = left.send(()); + } + Event::Frame { kind, body, .. } => { + let _ = frame.send((kind, body.to_vec())); + } + _ => {} + }, + ) + .unwrap(); + let first = openings.recv_timeout(Duration::from_secs(10)).unwrap(); + assert!(matches!( + host_met.try_recv(), + Err(mpsc::TryRecvError::Empty) + )); + assert!(matches!( + guest_met.try_recv(), + Err(mpsc::TryRecvError::Empty) + )); + let fresh = openings.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_ne!(first, fresh, "the retry reused the abandoned key exchange"); + let host_line = host_met.recv_timeout(Duration::from_secs(10)).unwrap(); + guest_met.recv_timeout(Duration::from_secs(10)).unwrap(); + let before_timeout = guest.shared.state.lock().unwrap().relay.clone().unwrap(); + guest + .shared + .state + .lock() + .unwrap() + .peers + .values() + .next() + .unwrap() + .pipe + .set_read_timeout(Duration::from_secs(1)) + .unwrap(); + host_line.send(kind::EDITS, &40_u64).unwrap(); + let (kind, body) = guest_frames.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!(kind, kind::EDITS); + assert_eq!(minicbor::decode::(&body).unwrap(), 40); + guest_left.recv_timeout(Duration::from_secs(10)).unwrap(); + let reopened = openings.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_ne!( + fresh, reopened, + "the retry reused the timed-out stream's keys" + ); + host_met.recv_timeout(Duration::from_secs(10)).unwrap(); + let line = guest_met.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!((host.failed(), guest.failed()), (0, 0)); + let before = guest.shared.state.lock().unwrap().relay.clone().unwrap(); + assert!(!Arc::ptr_eq(&before_timeout, &before)); + for edit in [41_u64, 42] { + line.send(kind::EDITS, &edit).unwrap(); + } + line.hang_up("left"); + let mut delivered = Vec::new(); + loop { + let (kind, body) = frames.recv_timeout(Duration::from_secs(10)).unwrap(); + match kind { + kind::EDITS => delivered.push(minicbor::decode::(&body).unwrap()), + kind::BYE => break, + _ => {} + } + } + assert_eq!(delivered, [41, 42]); + guest_left.recv_timeout(Duration::from_secs(10)).unwrap(); + host.set_presence(caret(43)); + until(&guest, |peers| { + peers + .first() + .is_some_and(|peer| peer.presence == Some(caret(43))) + }); + let state = guest.shared.state.lock().unwrap(); + assert!(Arc::ptr_eq(&before, state.relay.as_ref().unwrap())); + drop(state); + assert!(matches!( + host_met.try_recv(), + Err(mpsc::TryRecvError::Empty) + )); + assert!(matches!( + guest_met.try_recv(), + Err(mpsc::TryRecvError::Empty) + )); + assert!(matches!( + guest_frames.try_recv(), + Err(mpsc::TryRecvError::Empty) + )); + drop(guest); + drop(host); + proxy.join().unwrap(); +} + /// The end sharing a code asks the relay to number it; the other types the whole code. One /// with the wrong words never meets, and the relay, told so by the end sharing, burns the /// code once too many have tried.