authorclover caruso <git@paperclover.net> 2026-10-05 23:26:50-07:00
committerclover caruso <git@paperclover.net> 2026-10-10 22:56:05-07:00
log34a0250c70ab90c59387f05aa7840d58535392a6
tree8eb8ad81a42d873ce7d9e5873995083526a0e3e0
parentad380521b1078a34e058c9a67b8df523c30dfbaf
signature Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU (~clover)

fix: retry native LAN Live Share transport failures

Fixes #101 Assisted-by: gpt-6.1-sol

2 files changed, 332 insertions(+), 20 deletions(-)

crates/notebook/src/live.rs+46-20
...@@ -72,6 +72,7 @@ struct State {...@@ -72,6 +72,7 @@ struct State {
72 /// Counts changes to `presence`, so a writer sends only the newest.72 /// Counts changes to `presence`, so a writer sends only the newest.
73 generation: u64,73 generation: u64,
74 peers: BTreeMap<[u8; 16], Link>,74 peers: BTreeMap<[u8; 16], Link>,
75 dialing: BTreeMap<SocketAddr, thread::Thread>,
75 /// The code others type, once known.76 /// The code others type, once known.
76 code: Option<String>,77 code: Option<String>,
77 /// The relay connection open now, to hang up on leaving.78 /// The relay connection open now, to hang up on leaving.
...@@ -324,8 +325,7 @@ impl Live {...@@ -324,8 +325,7 @@ impl Live {
324325
325 /// Connects to a peer at `address` that discovery did not find.326 /// Connects to a peer at `address` that discovery did not find.
326 pub fn connect(&self, address: SocketAddr) {327 pub fn connect(&self, address: SocketAddr) {
327 let shared = Arc::clone(&self.shared);328 self.shared.dial(address);
328 thread::spawn(move || shared.dial(address));
329 }329 }
330330
331 /// Says where this end is now; peers hear only the newest of quick changes, at most every331 /// 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 {...@@ -424,6 +424,9 @@ impl Drop for Live {
424 // Wakes the accepting thread to see it has stopped.424 // Wakes the accepting thread to see it has stopped.
425 let _ = TcpStream::connect_timeout(&self.address, OPENING);425 let _ = TcpStream::connect_timeout(&self.address, OPENING);
426 let mut state = self.shared.state.lock().unwrap();426 let mut state = self.shared.state.lock().unwrap();
427 for dial in state.dialing.values() {
428 dial.unpark();
429 }
427 if let Some(daemon) = state.daemon.take() {430 if let Some(daemon) = state.daemon.take() {
428 let _ = daemon.shutdown();431 let _ = daemon.shutdown();
429 }432 }
...@@ -501,22 +504,45 @@ impl Shared {...@@ -501,22 +504,45 @@ impl Shared {
501 state.peers.keys().any(|id| hex(id) == peer)504 state.peers.keys().any(|id| hex(id) == peer)
502 }505 }
503506
504 /// Connects to `address`, and again while the peer is there and the connection was the507 /// Keeps one dial to `address`, meeting again after a transport failure.
505 /// one this end kept.508 fn dial(self: &Arc<Self>, address: SocketAddr) {
506 fn dial(self: Arc<Self>, address: SocketAddr) {509 let mut state = self.state.lock().unwrap();
507 let mut wait = Duration::from_secs(1);510 if self.stopped.load(Ordering::Acquire)
508 while let Ok(stream) = TcpStream::connect_timeout(&address, OPENING) {511 || state.burned
509 let Some(tag) = self.tag() else {512 || state.dialing.contains_key(&address)
510 return;513 {
511 };514 return;
512 if !Arc::clone(&self).run(Arc::new(stream), Side::Initiator, &tag)
513 || self.stopped.load(Ordering::Acquire)
514 {
515 return;
516 }
517 thread::sleep(wait);
518 wait = (wait * 2).min(PATIENCE);
519 }515 }
516 let shared = Arc::downgrade(self);
517 let dialing = thread::spawn(move || {
518 let mut wait = Duration::from_secs(1);
519 loop {
520 let Some(live) = shared.upgrade() else {
521 return;
522 };
523 if live.stopped.load(Ordering::Acquire) || live.state.lock().unwrap().burned {
524 break;
525 }
526 let Some(tag) = live.tag() else {
527 break;
528 };
529 drop(live);
530 if let Ok(stream) = TcpStream::connect_timeout(&address, OPENING) {
531 let Some(live) = shared.upgrade() else {
532 return;
533 };
534 if !live.run(Arc::new(stream), Side::Initiator, &tag) {
535 break;
536 }
537 }
538 thread::park_timeout(wait);
539 wait = (wait * 2).min(PATIENCE);
540 }
541 if let Some(live) = shared.upgrade() {
542 live.state.lock().unwrap().dialing.remove(&address);
543 }
544 });
545 state.dialing.insert(address, dialing.thread().clone());
520 }546 }
521547
522 /// Counts a meeting that failed on the secret: the end sharing a code burns it after too548 /// Counts a meeting that failed on the secret: the end sharing a code burns it after too
...@@ -534,7 +560,7 @@ impl Shared {...@@ -534,7 +560,7 @@ impl Shared {
534 }560 }
535561
536 /// Meets the peer at the other end of `pipe` in room `tag`, then reads from it until it562 /// Meets the peer at the other end of `pipe` in room `tag`, then reads from it until it
537 /// goes: whether it was the connection kept to that peer.563 /// goes: whether the next dial may try again.
538 fn run(self: Arc<Self>, pipe: Arc<dyn Pipe>, side: Side, tag: &str) -> bool {564 fn run(self: Arc<Self>, pipe: Arc<dyn Pipe>, side: Side, tag: &str) -> bool {
539 if self.stopped.load(Ordering::Acquire) || self.state.lock().unwrap().burned {565 if self.stopped.load(Ordering::Acquire) || self.state.lock().unwrap().burned {
540 pipe.shutdown();566 pipe.shutdown();
...@@ -579,7 +605,7 @@ impl Shared {...@@ -579,7 +605,7 @@ impl Shared {
579 pipe.broken();605 pipe.broken();
580 }606 }
581 pipe.shutdown();607 pipe.shutdown();
582 return false;608 return other.is_none() && error.kind() != io::ErrorKind::InvalidData;
583 }609 }
584 };610 };
585 let peer = hello.peer;611 let peer = hello.peer;
...@@ -767,7 +793,7 @@ fn discover(...@@ -767,7 +793,7 @@ fn discover(
767 addresses.sort_by_key(|ip| (!ip.is_ipv4(), !ip.is_loopback()));793 addresses.sort_by_key(|ip| (!ip.is_ipv4(), !ip.is_loopback()));
768 if let Some(ip) = addresses.first() {794 if let Some(ip) = addresses.first() {
769 let address = SocketAddr::new(*ip, service.port);795 let address = SocketAddr::new(*ip, service.port);
770 thread::spawn(move || shared.dial(address));796 shared.dial(address);
771 }797 }
772 }798 }
773 })799 })
crates/notebook/src/live/tests.rs+286
...@@ -70,6 +70,292 @@ fn peers_meet_and_follow_presence() {...@@ -70,6 +70,292 @@ fn peers_meet_and_follow_presence() {
70 until(&grace, <[Peer]>::is_empty);70 until(&grace, <[Peer]>::is_empty);
71}71}
7272
73fn lan_accept(listener: &TcpListener) -> TcpStream {
74 let deadline = Instant::now() + Duration::from_secs(10);
75 loop {
76 match listener.accept() {
77 Ok((stream, _)) => {
78 stream.set_nonblocking(false).unwrap();
79 stream.set_read_timeout(Some(OPENING)).unwrap();
80 stream.set_write_timeout(Some(OPENING)).unwrap();
81 return stream;
82 }
83 Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
84 assert!(Instant::now() < deadline, "LAN did not reconnect");
85 thread::sleep(Duration::from_millis(10));
86 }
87 Err(error) => panic!("{error}"),
88 }
89 }
90}
91
92fn lan_opening(stream: &mut TcpStream) -> Vec<u8> {
93 let mut length = [0; 4];
94 stream.read_exact(&mut length).unwrap();
95 let length = u32::from_be_bytes(length) as usize;
96 assert!(length < 1 << 20);
97 let mut opening = vec![0; length];
98 stream.read_exact(&mut opening).unwrap();
99 opening
100}
101
102#[test]
103fn lan_retries_opening_io_and_reconnect_refusal_with_fresh_keys() {
104 let room = Room::Notebook([71; 16]);
105 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
106 let address = listener.local_addr().unwrap();
107 listener.set_nonblocking(true).unwrap();
108 let (fault, faults) = mpsc::channel();
109 let (resume, resumed) = mpsc::channel();
110 let server_room = room.clone();
111 let server = thread::spawn(move || {
112 let mut abandoned = lan_accept(&listener);
113 let first = lan_opening(&mut abandoned);
114 drop(abandoned);
115 let host = hello("Host");
116 let mut openings = vec![first];
117 let mut delivered = Vec::new();
118 let mut listener = Some(listener);
119 for edit in [41_u64, 42] {
120 let mut stream = lan_accept(listener.as_ref().unwrap());
121 let theirs = lan_opening(&mut stream);
122 assert!(openings.iter().all(|old| *old != theirs), "LAN reused keys");
123 openings.push(theirs.clone());
124 let (opening, ours) = wire::Opening::new(
125 Side::Responder,
126 &server_room.tag().unwrap(),
127 &server_room.secret(),
128 )
129 .unwrap();
130 stream
131 .write_all(&(ours.len() as u32).to_be_bytes())
132 .unwrap();
133 stream.write_all(&ours).unwrap();
134 let (mut send, mut receive) = opening.finish(&theirs).unwrap();
135 send.send(&mut stream, kind::HELLO, &host).unwrap();
136 assert_eq!(receive.receive(&mut stream).unwrap().0, kind::HELLO);
137 send.send(&mut stream, kind::EDITS, &edit).unwrap();
138 loop {
139 let (kind, body) = receive.receive(&mut stream).unwrap();
140 if kind == kind::EDITS {
141 delivered.push(minicbor::decode::<u64>(&body).unwrap());
142 break;
143 }
144 }
145 assert_eq!(
146 listener.as_ref().unwrap().accept().unwrap_err().kind(),
147 io::ErrorKind::WouldBlock,
148 "duplicate LAN dial workers connected",
149 );
150 if edit == 41 {
151 drop(listener.take());
152 stream.shutdown(Shutdown::Both).unwrap();
153 fault.send(()).unwrap();
154 resumed.recv_timeout(Duration::from_secs(10)).unwrap();
155 let reopened = TcpListener::bind(address).unwrap();
156 reopened.set_nonblocking(true).unwrap();
157 listener = Some(reopened);
158 } else {
159 loop {
160 let (kind, _) = receive.receive(&mut stream).unwrap();
161 assert_ne!(kind, kind::EDITS, "an edit was delivered twice");
162 if kind == kind::BYE {
163 break;
164 }
165 }
166 }
167 }
168 delivered
169 });
170 let (frame, frames) = mpsc::channel();
171 let guest = Live::start(hello("Guest"), &room, None, None, move |event| {
172 if let Event::Frame { kind, body, .. } = event {
173 frame.send((kind, body.to_vec())).unwrap();
174 }
175 })
176 .unwrap();
177 for _ in 0..20 {
178 guest.connect(address);
179 }
180 for edit in [41_u64, 42] {
181 let (kind, body) = frames.recv_timeout(Duration::from_secs(10)).unwrap();
182 assert_eq!(kind, kind::EDITS);
183 assert_eq!(minicbor::decode::<u64>(&body).unwrap(), edit);
184 let peer = until(&guest, |peers| peers.len() == 1).remove(0);
185 let line = guest.line(&peer.hello.peer).unwrap();
186 line.send(kind::EDITS, &edit).unwrap();
187 if edit == 41 {
188 faults.recv_timeout(Duration::from_secs(10)).unwrap();
189 until(&guest, <[Peer]>::is_empty);
190 assert_eq!(guest.failed(), 0);
191 assert_eq!(
192 TcpStream::connect(address).unwrap_err().kind(),
193 io::ErrorKind::ConnectionRefused
194 );
195 thread::sleep(Duration::from_millis(2300));
196 resume.send(()).unwrap();
197 } else {
198 line.hang_up("left");
199 }
200 }
201 assert_eq!(server.join().unwrap(), [41, 42]);
202 assert!(matches!(frames.try_recv(), Err(mpsc::TryRecvError::Empty)));
203 assert_eq!(guest.failed(), 0);
204 until(&guest, <[Peer]>::is_empty);
205 let stopped = Arc::downgrade(&guest.shared);
206 drop(guest);
207 let deadline = Instant::now() + Duration::from_secs(1);
208 while stopped.upgrade().is_some() {
209 assert!(
210 Instant::now() < deadline,
211 "LAN helpers retained stopped Live"
212 );
213 thread::sleep(Duration::from_millis(10));
214 }
215}
216
217#[test]
218fn lan_retries_initial_refusal_and_stops_its_backoff() {
219 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
220 let address = listener.local_addr().unwrap();
221 drop(listener);
222 let room = Room::Notebook([72; 16]);
223 let guest = Live::start(hello("Guest"), &room, None, None, |_| {}).unwrap();
224 guest.connect(address);
225 thread::sleep(Duration::from_millis(1200));
226 let listener = TcpListener::bind(address).unwrap();
227 listener.set_nonblocking(true).unwrap();
228 let mut stream = lan_accept(&listener);
229 let first = lan_opening(&mut stream);
230 assert!(!first.is_empty());
231 drop(listener);
232 drop(stream);
233 thread::sleep(Duration::from_millis(100));
234 let stopped = Arc::downgrade(&guest.shared);
235 drop(guest);
236 let deadline = Instant::now() + Duration::from_secs(1);
237 while stopped.upgrade().is_some() {
238 assert!(
239 Instant::now() < deadline,
240 "LAN backoff retained stopped Live"
241 );
242 thread::sleep(Duration::from_millis(10));
243 }
244 let listener = TcpListener::bind(address).unwrap();
245 listener.set_nonblocking(true).unwrap();
246 thread::sleep(Duration::from_millis(2200));
247 assert_eq!(
248 listener.accept().unwrap_err().kind(),
249 io::ErrorKind::WouldBlock
250 );
251}
252
253#[test]
254fn lan_authentication_and_version_failures_stop_retrying() {
255 for version in [false, true] {
256 let room = Room::Notebook([73; 16]);
257 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
258 listener.set_nonblocking(true).unwrap();
259 let address = listener.local_addr().unwrap();
260 let guest = Live::start(hello("Guest"), &room, None, None, |_| {}).unwrap();
261 guest.connect(address);
262 let mut stream = lan_accept(&listener);
263 if version {
264 lan_opening(&mut stream);
265 let mut opening = minicbor::Encoder::new(Vec::new());
266 opening
267 .map(3)
268 .unwrap()
269 .u8(0)
270 .unwrap()
271 .u16(wire::VERSION + 1)
272 .unwrap()
273 .u8(1)
274 .unwrap()
275 .str(&room.tag().unwrap())
276 .unwrap()
277 .u8(2)
278 .unwrap()
279 .bytes(&[0; 33])
280 .unwrap();
281 let opening = opening.into_writer();
282 stream
283 .write_all(&(opening.len() as u32).to_be_bytes())
284 .unwrap();
285 stream.write_all(&opening).unwrap();
286 until(&guest, |_| guest.other_version() == Some(wire::VERSION + 1));
287 assert_eq!(guest.failed(), 0);
288 } else {
289 let (mut send, _) = wire::open(
290 &mut stream,
291 Side::Responder,
292 &room.tag().unwrap(),
293 b"wrong secret",
294 )
295 .unwrap();
296 send.send(&mut stream, kind::HELLO, &hello("Host")).unwrap();
297 until(&guest, |_| guest.failed() == 1);
298 }
299 drop(stream);
300 thread::sleep(Duration::from_millis(1200));
301 assert_eq!(
302 listener.accept().unwrap_err().kind(),
303 io::ErrorKind::WouldBlock
304 );
305 assert_eq!(guest.failed(), u32::from(!version));
306 guest.connect(address);
307 let mut stream = lan_accept(&listener);
308 assert!(
309 !lan_opening(&mut stream).is_empty(),
310 "finished dial was never released"
311 );
312 drop(guest);
313 }
314}
315
316#[test]
317fn lan_duplicate_peer_stops_retrying_without_replacing_its_line() {
318 let room = Room::Notebook([74; 16]);
319 let host_hello = hello("Host");
320 let (frame, frames) = mpsc::channel();
321 let host = Live::start(host_hello.clone(), &room, None, None, move |event| {
322 if let Event::Frame { kind, body, .. } = event {
323 frame.send((kind, body.to_vec())).unwrap();
324 }
325 })
326 .unwrap();
327 let guest = Live::start(hello("Guest"), &room, None, None, |_| {}).unwrap();
328 guest.connect(host.address());
329 until(&guest, |peers| peers.len() == 1);
330 let line = guest.line(&host_hello.peer).unwrap();
331 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
332 listener.set_nonblocking(true).unwrap();
333 guest.connect(listener.local_addr().unwrap());
334 let mut stream = lan_accept(&listener);
335 let (mut send, mut receive) = wire::open(
336 &mut stream,
337 Side::Responder,
338 &room.tag().unwrap(),
339 &room.secret(),
340 )
341 .unwrap();
342 send.send(&mut stream, kind::HELLO, &host_hello).unwrap();
343 assert_eq!(receive.receive(&mut stream).unwrap().0, kind::HELLO);
344 assert_eq!(stream.read(&mut [0; 1]).unwrap(), 0);
345 thread::sleep(Duration::from_millis(1200));
346 assert_eq!(
347 listener.accept().unwrap_err().kind(),
348 io::ErrorKind::WouldBlock
349 );
350 assert_eq!(guest.peers().len(), 1);
351 assert_eq!(guest.failed(), 0);
352 line.send(kind::EDITS, &43_u64).unwrap();
353 let (kind, body) = frames.recv_timeout(Duration::from_secs(10)).unwrap();
354 assert_eq!(kind, kind::EDITS);
355 assert_eq!(minicbor::decode::<u64>(&body).unwrap(), 43);
356 assert!(matches!(frames.try_recv(), Err(mpsc::TryRecvError::Empty)));
357}
358
73/// A peer holding another code never meets: its first frame does not open.359/// A peer holding another code never meets: its first frame does not open.
74#[test]360#[test]
75fn another_code_never_meets() {361fn another_code_never_meets() {