diff --git a/crates/notebook/src/live.rs b/crates/notebook/src/live.rs index f2b4fc7714d02a253c92f156b5f6cfe8620c24f3..69906219840dc6e2a84930fbba5ea4f98f7f6513 100644 --- a/crates/notebook/src/live.rs +++ b/crates/notebook/src/live.rs @@ -72,6 +72,7 @@ struct State { /// Counts changes to `presence`, so a writer sends only the newest. generation: u64, peers: BTreeMap<[u8; 16], Link>, + dialing: BTreeMap, /// The code others type, once known. code: Option, /// The relay connection open now, to hang up on leaving. @@ -324,8 +325,7 @@ impl Live { /// Connects to a peer at `address` that discovery did not find. pub fn connect(&self, address: SocketAddr) { - let shared = Arc::clone(&self.shared); - thread::spawn(move || shared.dial(address)); + self.shared.dial(address); } /// Says where this end is now; peers hear only the newest of quick changes, at most every @@ -424,6 +424,9 @@ impl Drop for Live { // Wakes the accepting thread to see it has stopped. let _ = TcpStream::connect_timeout(&self.address, OPENING); let mut state = self.shared.state.lock().unwrap(); + for dial in state.dialing.values() { + dial.unpark(); + } if let Some(daemon) = state.daemon.take() { let _ = daemon.shutdown(); } @@ -501,22 +504,45 @@ impl Shared { state.peers.keys().any(|id| hex(id) == peer) } - /// Connects to `address`, and again while the peer is there and the connection was the - /// one this end kept. - fn dial(self: Arc, address: SocketAddr) { - let mut wait = Duration::from_secs(1); - while let Ok(stream) = TcpStream::connect_timeout(&address, OPENING) { - let Some(tag) = self.tag() else { - return; - }; - if !Arc::clone(&self).run(Arc::new(stream), Side::Initiator, &tag) - || self.stopped.load(Ordering::Acquire) - { - return; - } - thread::sleep(wait); - wait = (wait * 2).min(PATIENCE); + /// Keeps one dial to `address`, meeting again after a transport failure. + fn dial(self: &Arc, address: SocketAddr) { + let mut state = self.state.lock().unwrap(); + if self.stopped.load(Ordering::Acquire) + || state.burned + || state.dialing.contains_key(&address) + { + return; } + let shared = Arc::downgrade(self); + let dialing = thread::spawn(move || { + let mut wait = Duration::from_secs(1); + loop { + let Some(live) = shared.upgrade() else { + return; + }; + if live.stopped.load(Ordering::Acquire) || live.state.lock().unwrap().burned { + break; + } + let Some(tag) = live.tag() else { + break; + }; + drop(live); + if let Ok(stream) = TcpStream::connect_timeout(&address, OPENING) { + let Some(live) = shared.upgrade() else { + return; + }; + if !live.run(Arc::new(stream), Side::Initiator, &tag) { + break; + } + } + thread::park_timeout(wait); + wait = (wait * 2).min(PATIENCE); + } + if let Some(live) = shared.upgrade() { + live.state.lock().unwrap().dialing.remove(&address); + } + }); + state.dialing.insert(address, dialing.thread().clone()); } /// Counts a meeting that failed on the secret: the end sharing a code burns it after too @@ -534,7 +560,7 @@ impl Shared { } /// Meets the peer at the other end of `pipe` in room `tag`, then reads from it until it - /// goes: whether it was the connection kept to that peer. + /// goes: whether the next dial may try again. fn run(self: Arc, pipe: Arc, side: Side, tag: &str) -> bool { if self.stopped.load(Ordering::Acquire) || self.state.lock().unwrap().burned { pipe.shutdown(); @@ -579,7 +605,7 @@ impl Shared { pipe.broken(); } pipe.shutdown(); - return false; + return other.is_none() && error.kind() != io::ErrorKind::InvalidData; } }; let peer = hello.peer; @@ -767,7 +793,7 @@ fn discover( addresses.sort_by_key(|ip| (!ip.is_ipv4(), !ip.is_loopback())); if let Some(ip) = addresses.first() { let address = SocketAddr::new(*ip, service.port); - thread::spawn(move || shared.dial(address)); + shared.dial(address); } } }) diff --git a/crates/notebook/src/live/tests.rs b/crates/notebook/src/live/tests.rs index 1befa640fb5d0e131f19eb79099088be29f3a65b..98876ac993ae662372bc9d290ff315b864f1cf56 100644 --- a/crates/notebook/src/live/tests.rs +++ b/crates/notebook/src/live/tests.rs @@ -70,6 +70,292 @@ fn peers_meet_and_follow_presence() { until(&grace, <[Peer]>::is_empty); } +fn lan_accept(listener: &TcpListener) -> TcpStream { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + match listener.accept() { + Ok((stream, _)) => { + stream.set_nonblocking(false).unwrap(); + stream.set_read_timeout(Some(OPENING)).unwrap(); + stream.set_write_timeout(Some(OPENING)).unwrap(); + return stream; + } + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + assert!(Instant::now() < deadline, "LAN did not reconnect"); + thread::sleep(Duration::from_millis(10)); + } + Err(error) => panic!("{error}"), + } + } +} + +fn lan_opening(stream: &mut TcpStream) -> Vec { + let mut length = [0; 4]; + stream.read_exact(&mut length).unwrap(); + let length = u32::from_be_bytes(length) as usize; + assert!(length < 1 << 20); + let mut opening = vec![0; length]; + stream.read_exact(&mut opening).unwrap(); + opening +} + +#[test] +fn lan_retries_opening_io_and_reconnect_refusal_with_fresh_keys() { + let room = Room::Notebook([71; 16]); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + listener.set_nonblocking(true).unwrap(); + let (fault, faults) = mpsc::channel(); + let (resume, resumed) = mpsc::channel(); + let server_room = room.clone(); + let server = thread::spawn(move || { + let mut abandoned = lan_accept(&listener); + let first = lan_opening(&mut abandoned); + drop(abandoned); + let host = hello("Host"); + let mut openings = vec![first]; + let mut delivered = Vec::new(); + let mut listener = Some(listener); + for edit in [41_u64, 42] { + let mut stream = lan_accept(listener.as_ref().unwrap()); + let theirs = lan_opening(&mut stream); + assert!(openings.iter().all(|old| *old != theirs), "LAN reused keys"); + openings.push(theirs.clone()); + let (opening, ours) = wire::Opening::new( + Side::Responder, + &server_room.tag().unwrap(), + &server_room.secret(), + ) + .unwrap(); + stream + .write_all(&(ours.len() as u32).to_be_bytes()) + .unwrap(); + stream.write_all(&ours).unwrap(); + let (mut send, mut receive) = opening.finish(&theirs).unwrap(); + send.send(&mut stream, kind::HELLO, &host).unwrap(); + assert_eq!(receive.receive(&mut stream).unwrap().0, kind::HELLO); + send.send(&mut stream, kind::EDITS, &edit).unwrap(); + loop { + let (kind, body) = receive.receive(&mut stream).unwrap(); + if kind == kind::EDITS { + delivered.push(minicbor::decode::(&body).unwrap()); + break; + } + } + assert_eq!( + listener.as_ref().unwrap().accept().unwrap_err().kind(), + io::ErrorKind::WouldBlock, + "duplicate LAN dial workers connected", + ); + if edit == 41 { + drop(listener.take()); + stream.shutdown(Shutdown::Both).unwrap(); + fault.send(()).unwrap(); + resumed.recv_timeout(Duration::from_secs(10)).unwrap(); + let reopened = TcpListener::bind(address).unwrap(); + reopened.set_nonblocking(true).unwrap(); + listener = Some(reopened); + } else { + loop { + let (kind, _) = receive.receive(&mut stream).unwrap(); + assert_ne!(kind, kind::EDITS, "an edit was delivered twice"); + if kind == kind::BYE { + break; + } + } + } + } + delivered + }); + let (frame, frames) = mpsc::channel(); + let guest = Live::start(hello("Guest"), &room, None, None, move |event| { + if let Event::Frame { kind, body, .. } = event { + frame.send((kind, body.to_vec())).unwrap(); + } + }) + .unwrap(); + for _ in 0..20 { + guest.connect(address); + } + for edit in [41_u64, 42] { + let (kind, body) = frames.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!(kind, kind::EDITS); + assert_eq!(minicbor::decode::(&body).unwrap(), edit); + let peer = until(&guest, |peers| peers.len() == 1).remove(0); + let line = guest.line(&peer.hello.peer).unwrap(); + line.send(kind::EDITS, &edit).unwrap(); + if edit == 41 { + faults.recv_timeout(Duration::from_secs(10)).unwrap(); + until(&guest, <[Peer]>::is_empty); + assert_eq!(guest.failed(), 0); + assert_eq!( + TcpStream::connect(address).unwrap_err().kind(), + io::ErrorKind::ConnectionRefused + ); + thread::sleep(Duration::from_millis(2300)); + resume.send(()).unwrap(); + } else { + line.hang_up("left"); + } + } + assert_eq!(server.join().unwrap(), [41, 42]); + assert!(matches!(frames.try_recv(), Err(mpsc::TryRecvError::Empty))); + assert_eq!(guest.failed(), 0); + until(&guest, <[Peer]>::is_empty); + let stopped = Arc::downgrade(&guest.shared); + drop(guest); + let deadline = Instant::now() + Duration::from_secs(1); + while stopped.upgrade().is_some() { + assert!( + Instant::now() < deadline, + "LAN helpers retained stopped Live" + ); + thread::sleep(Duration::from_millis(10)); + } +} + +#[test] +fn lan_retries_initial_refusal_and_stops_its_backoff() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + drop(listener); + let room = Room::Notebook([72; 16]); + let guest = Live::start(hello("Guest"), &room, None, None, |_| {}).unwrap(); + guest.connect(address); + thread::sleep(Duration::from_millis(1200)); + let listener = TcpListener::bind(address).unwrap(); + listener.set_nonblocking(true).unwrap(); + let mut stream = lan_accept(&listener); + let first = lan_opening(&mut stream); + assert!(!first.is_empty()); + drop(listener); + drop(stream); + thread::sleep(Duration::from_millis(100)); + let stopped = Arc::downgrade(&guest.shared); + drop(guest); + let deadline = Instant::now() + Duration::from_secs(1); + while stopped.upgrade().is_some() { + assert!( + Instant::now() < deadline, + "LAN backoff retained stopped Live" + ); + thread::sleep(Duration::from_millis(10)); + } + let listener = TcpListener::bind(address).unwrap(); + listener.set_nonblocking(true).unwrap(); + thread::sleep(Duration::from_millis(2200)); + assert_eq!( + listener.accept().unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); +} + +#[test] +fn lan_authentication_and_version_failures_stop_retrying() { + for version in [false, true] { + let room = Room::Notebook([73; 16]); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let address = listener.local_addr().unwrap(); + let guest = Live::start(hello("Guest"), &room, None, None, |_| {}).unwrap(); + guest.connect(address); + let mut stream = lan_accept(&listener); + if version { + lan_opening(&mut stream); + let mut opening = minicbor::Encoder::new(Vec::new()); + opening + .map(3) + .unwrap() + .u8(0) + .unwrap() + .u16(wire::VERSION + 1) + .unwrap() + .u8(1) + .unwrap() + .str(&room.tag().unwrap()) + .unwrap() + .u8(2) + .unwrap() + .bytes(&[0; 33]) + .unwrap(); + let opening = opening.into_writer(); + stream + .write_all(&(opening.len() as u32).to_be_bytes()) + .unwrap(); + stream.write_all(&opening).unwrap(); + until(&guest, |_| guest.other_version() == Some(wire::VERSION + 1)); + assert_eq!(guest.failed(), 0); + } else { + let (mut send, _) = wire::open( + &mut stream, + Side::Responder, + &room.tag().unwrap(), + b"wrong secret", + ) + .unwrap(); + send.send(&mut stream, kind::HELLO, &hello("Host")).unwrap(); + until(&guest, |_| guest.failed() == 1); + } + drop(stream); + thread::sleep(Duration::from_millis(1200)); + assert_eq!( + listener.accept().unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + assert_eq!(guest.failed(), u32::from(!version)); + guest.connect(address); + let mut stream = lan_accept(&listener); + assert!( + !lan_opening(&mut stream).is_empty(), + "finished dial was never released" + ); + drop(guest); + } +} + +#[test] +fn lan_duplicate_peer_stops_retrying_without_replacing_its_line() { + let room = Room::Notebook([74; 16]); + let host_hello = hello("Host"); + let (frame, frames) = mpsc::channel(); + let host = Live::start(host_hello.clone(), &room, None, None, move |event| { + if let Event::Frame { kind, body, .. } = event { + frame.send((kind, body.to_vec())).unwrap(); + } + }) + .unwrap(); + let guest = Live::start(hello("Guest"), &room, None, None, |_| {}).unwrap(); + guest.connect(host.address()); + until(&guest, |peers| peers.len() == 1); + let line = guest.line(&host_hello.peer).unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + guest.connect(listener.local_addr().unwrap()); + let mut stream = lan_accept(&listener); + let (mut send, mut receive) = wire::open( + &mut stream, + Side::Responder, + &room.tag().unwrap(), + &room.secret(), + ) + .unwrap(); + send.send(&mut stream, kind::HELLO, &host_hello).unwrap(); + assert_eq!(receive.receive(&mut stream).unwrap().0, kind::HELLO); + assert_eq!(stream.read(&mut [0; 1]).unwrap(), 0); + thread::sleep(Duration::from_millis(1200)); + assert_eq!( + listener.accept().unwrap_err().kind(), + io::ErrorKind::WouldBlock + ); + assert_eq!(guest.peers().len(), 1); + assert_eq!(guest.failed(), 0); + line.send(kind::EDITS, &43_u64).unwrap(); + let (kind, body) = frames.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!(kind, kind::EDITS); + assert_eq!(minicbor::decode::(&body).unwrap(), 43); + assert!(matches!(frames.try_recv(), Err(mpsc::TryRecvError::Empty))); +} + /// A peer holding another code never meets: its first frame does not open. #[test] fn another_code_never_meets() {