authorclover caruso <git@paperclover.net> 2026-10-15 01:54:47-07:00
committerclover caruso <git@paperclover.net> 2026-10-15 02:18:54-07:00
log6bf6a20c4073e708ed209be99349432815b693ce
treee490bdd038655b4ed9fd7cdd934c04c4e595da65
parent0c2795ab06a608463deba57402a39ada750c526c
signature Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU (~clover)

fix(live): deliver replies from retired routes

A REPLY read off a route that was just retired reaches the waiter matching its id instead of leaving it to time out as lost. - Allow drops a request whose route has gone instead of keeping it silently. - Presence frames check their one sender rather than building the peer list. - The guest joins presence in one step; the WELCOME fence covers the pre-check. - The browser Live names its field like the native one, dropping cfg pairs. - Native removal sends welcomes without the browser-only rotation guard. - Handover tests live in one module, admit guests through Served::admit, wait on reader channels, and print no evidence markers. Refs #141 Assisted-by: claude-opus-5.5

5 files changed, 327 insertions(+), 392 deletions(-)

crates/notebook/src/live.rs+9-5
......@@ -455,15 +455,19 @@ impl Shared {
455455 !self.state.lock().unwrap().peers.contains_key(&hello.peer)
456456 }
457457
458 /// Whether `hello` is how a peer is in the room now: its link, or its place in the group.
458459 fn connected(&self, hello: &Arc<Hello>) -> bool {
460 let current = |peer: &Peer| Arc::ptr_eq(&peer.hello, hello);
461 let state = self.state.lock().unwrap();
459462 !self.stopped.load(Ordering::Acquire)
460 && self
461 .state
462 .lock()
463 .unwrap()
463 && (state
464464 .peers
465465 .get(&hello.peer)
466 .is_some_and(|link| Arc::ptr_eq(&link.peer.hello, hello))
466 .is_some_and(|link| current(&link.peer))
467 || state.group().is_some_and(|group| {
468 (group.members.lock().unwrap().values())
469 .any(|member| member.peer.as_ref().is_some_and(current))
470 }))
467471 }
468472
469473 /// The peers in the room: those met directly, and the rest as the relay's group or a
crates/notebook/src/live/share.rs+25-54
......@@ -29,9 +29,6 @@ use web_time::Instant;
2929
3030#[path = "share/batch.rs"]
3131mod batch;
32#[cfg(all(test, not(target_arch = "wasm32")))]
33#[path = "share/handover_tests.rs"]
34mod handover_tests;
3532#[path = "share/membership.rs"]
3633mod membership;
3734#[cfg(target_arch = "wasm32")]
......@@ -1643,6 +1640,21 @@ impl Inner {
16431640 reports.lost();
16441641 }
16451642 }
1643 // A reply only reaches its waiter, so one read off a retired route still counts.
1644 Event::Frame {
1645 from,
1646 kind: kind::REPLY,
1647 body,
1648 } => {
1649 let serves = (self.host.lock().unwrap().as_ref())
1650 .is_some_and(|(host, _)| host.peer == from.peer);
1651 if serves && let Ok(reply) = minicbor::decode::<Reply>(body) {
1652 let waiting = self.pending.lock().unwrap().remove(&reply.id);
1653 if let Some(waiting) = waiting {
1654 let _ = waiting.send(reply);
1655 }
1656 }
1657 }
16461658 Event::Frame { from, kind, body } => {
16471659 let host = self.host.lock().unwrap();
16481660 if !host
......@@ -1654,20 +1666,14 @@ impl Inner {
16541666 }
16551667 match kind {
16561668 kind::WELCOME => {
1657 if let Ok(welcome) = minicbor::decode::<Welcome>(body) {
1669 if let Ok(welcome) = minicbor::decode::<Welcome>(body)
1670 && !(self.presence.lock().unwrap().as_ref())
1671 .is_some_and(|(held, _)| *held == welcome.room)
1672 {
16581673 drop(host);
16591674 self.join_presence(source, from, welcome.room);
16601675 }
16611676 }
1662 kind::REPLY => {
1663 if let Ok(reply) = minicbor::decode::<Reply>(body) {
1664 let waiting = self.pending.lock().unwrap().remove(&reply.id);
1665 drop(host);
1666 if let Some(waiting) = waiting {
1667 let _ = waiting.send(reply);
1668 }
1669 }
1670 }
16711677 kind::DELTA => {
16721678 if let Ok(delta) = minicbor::decode::<Delta>(body) {
16731679 self.apply(&delta);
......@@ -1708,22 +1714,6 @@ impl Inner {
17081714 from: &Arc<Hello>,
17091715 secret: [u8; 16],
17101716 ) {
1711 {
1712 let host = self.host.lock().unwrap();
1713 if !host
1714 .as_ref()
1715 .is_some_and(|(admitted, _)| Arc::ptr_eq(admitted, from))
1716 || !source.connected(from)
1717 || self
1718 .presence
1719 .lock()
1720 .unwrap()
1721 .as_ref()
1722 .is_some_and(|(held, _)| *held == secret)
1723 {
1724 return;
1725 }
1726 }
17271717 let inner = Arc::downgrade(self);
17281718 let connected = Mutex::new(None);
17291719 let live = Live::start_with(
......@@ -1741,11 +1731,7 @@ impl Inner {
17411731 let Some((_, live)) = presence.as_ref() else {
17421732 return;
17431733 };
1744 #[cfg(not(target_arch = "wasm32"))]
1745 let selected = Arc::ptr_eq(&live.shared, source);
1746 #[cfg(target_arch = "wasm32")]
1747 let selected = Arc::ptr_eq(&live.0, source);
1748 if !selected {
1734 if !Arc::ptr_eq(&live.shared, source) {
17491735 return;
17501736 }
17511737 match event {
......@@ -1776,10 +1762,7 @@ impl Inner {
17761762 if host
17771763 .as_ref()
17781764 .is_some_and(|(host, _)| host.peer == from.peer)
1779 && source
1780 .peers()
1781 .iter()
1782 .any(|peer| Arc::ptr_eq(&peer.hello, from)) =>
1765 && source.connected(from) =>
17831766 {
17841767 match kind {
17851768 kind::DELTA => {
......@@ -1813,25 +1796,13 @@ impl Inner {
18131796 (inner.events)();
18141797 },
18151798 );
1816 match live {
1817 Ok(live) => self.install_presence(source, from, secret, live),
1818 Err(error) => eprintln!("Live Share: could not join presence: {error}"),
1819 }
1820 }
1821
1822 fn install_presence(
1823 self: &Arc<Self>,
1824 source: &Arc<super::Shared>,
1825 from: &Arc<Hello>,
1826 secret: [u8; 16],
1827 live: Live,
1828 ) {
1799 let live = match live {
1800 Ok(live) => live,
1801 Err(error) => return eprintln!("Live Share: could not join presence: {error}"),
1802 };
18291803 let here = self.here.lock().unwrap();
18301804 live.set_presence(here.clone());
1831 #[cfg(not(target_arch = "wasm32"))]
18321805 let refresh = Arc::clone(&live.shared);
1833 #[cfg(target_arch = "wasm32")]
1834 let refresh = Arc::clone(&live.0);
18351806 let mut candidate = Some(live);
18361807 let (retired, paths, reports) = {
18371808 let host = self.host.lock().unwrap();
crates/notebook/src/live/share/handover_tests.rs+236-117
......@@ -1,8 +1,9 @@
1use super::super::{GONE, OPENING, Pipe, Shared, State};
1use super::super::super::{GONE, OPENING, Pipe, Shared, State};
22use super::*;
33use std::{
44 io::{Read, Write},
55 net::{Shutdown, TcpListener, TcpStream},
6 sync::atomic::AtomicUsize,
67};
78
89struct Barrier {
......@@ -144,7 +145,7 @@ impl Handover {
144145 }
145146 }
146147
147 fn open(&mut self, host: &Hello, direct: bool) -> (TcpStream, wire::Sealer) {
148 fn open(&mut self, host: &Hello, direct: bool) -> (TcpStream, wire::Sealer, wire::Sealer) {
148149 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
149150 let mut remote =
150151 TcpStream::connect_timeout(&listener.local_addr().unwrap(), OPENING).unwrap();
......@@ -180,7 +181,7 @@ impl Handover {
180181 let (message, body) = receive.receive(&mut remote).unwrap();
181182 assert_eq!(message, kind::HELLO);
182183 assert_eq!(minicbor::decode::<Hello>(&body).unwrap(), self.shared.me);
183 (remote, send)
184 (remote, send, receive)
184185 }
185186
186187 fn heard(&self) {
......@@ -198,25 +199,17 @@ impl Drop for Handover {
198199 pipe.shutdown();
199200 }
200201 self.shared.state.lock().unwrap().peers.clear();
201 let deadline = std::time::Instant::now() + OPENING;
202 for (_, reader) in &self.readers {
203 while !reader.is_finished() && std::time::Instant::now() < deadline {
204 thread::yield_now();
205 }
206 }
207 let readers_done = self.readers.iter().all(|(_, reader)| reader.is_finished());
202 let readers_done = self
203 .readers
204 .iter()
205 .all(|(done, _)| done.recv_timeout(OPENING) != Err(mpsc::RecvTimeoutError::Timeout));
208206 self.guest.host.lock().unwrap().take();
209207 let count = self.pipes.len();
210208 self.pipes.clear();
211 let retired = (0..count).all(|_| self.retired.1.recv_timeout(OPENING).is_ok());
212 let mut joined = readers_done && retired;
213 if readers_done {
214 for (_, reader) in self.readers.drain(..) {
215 joined &= reader.join().is_ok();
216 }
217 }
209 let joined = readers_done
210 && (0..count).all(|_| self.retired.1.recv_timeout(OPENING).is_ok())
211 && (self.readers.drain(..)).all(|(_, reader)| reader.join().is_ok());
218212 if !thread::panicking() {
219 println!("HANDOVER_READERS_JOINED_AND_PIPE_OWNERS_RETIRED={joined};PIPES={count}");
220213 assert!(
221214 joined,
222215 "handover reader/writer owners did not retire and join"
......@@ -230,7 +223,7 @@ fn handover_stale_frame_cannot_end_promoted_guest() {
230223 let mut fixture = Handover::new(None);
231224 let mut host = Hello::new("Old host route".into(), None).unwrap();
232225 host.serves = Some(fixture.guest.share);
233 let (mut old, mut send) = fixture.open(&host, false);
226 let (mut old, mut send, _) = fixture.open(&host, false);
234227 fixture.heard();
235228 let (gate, copied, resume) = barrier();
236229 fixture.releases.push(resume.clone());
......@@ -245,7 +238,7 @@ fn handover_stale_frame_cannot_end_promoted_guest() {
245238 .unwrap();
246239 copied.recv_timeout(OPENING).unwrap();
247240 host.name = "Promoted host route".into();
248 let (_new, _) = fixture.open(&host, true);
241 let (_new, _, _) = fixture.open(&host, true);
249242 fixture.heard();
250243 assert!(
251244 fixture.shared.state.lock().unwrap().peers[&host.peer]
......@@ -257,10 +250,42 @@ fn handover_stale_frame_cannot_end_promoted_guest() {
257250 fixture.readers[0].0.recv_timeout(OPENING).unwrap();
258251 let ended = *fixture.guest.ended.lock().unwrap();
259252 drop(fixture);
260 println!("HANDOVER_STALE_FRAME_ENDED={ended:?}");
261253 assert_eq!(ended, None, "stale frame ended the promoted guest");
262254}
263255
256#[test]
257fn handover_reply_from_retired_route_reaches_its_waiter() {
258 let mut fixture = Handover::new(None);
259 let mut host = Hello::new("Old host route".into(), None).unwrap();
260 host.serves = Some(fixture.guest.share);
261 let (mut old, mut send, _) = fixture.open(&host, false);
262 fixture.heard();
263 let (answer, waiting) = crate::task::response::<Reply>();
264 fixture.guest.pending.lock().unwrap().insert(75, answer);
265 let (gate, copied, resume) = barrier();
266 fixture.releases.push(resume.clone());
267 *fixture.pipes[0].body.lock().unwrap() = Some(gate);
268 let reply = Reply {
269 id: 75,
270 length: Some(12),
271 ..Reply::default()
272 };
273 send.send(&mut old, kind::REPLY, &reply).unwrap();
274 copied.recv_timeout(OPENING).unwrap();
275 host.name = "Promoted host route".into();
276 let (_new, _, _) = fixture.open(&host, true);
277 fixture.heard();
278 resume.send(()).unwrap();
279 fixture.readers[0].0.recv_timeout(OPENING).unwrap();
280 let answered = waiting.try_recv().ok();
281 drop(fixture);
282 assert_eq!(
283 answered,
284 Some(reply),
285 "a reply read off a retired route missed its waiter"
286 );
287}
288
264289#[test]
265290fn handover_stale_met_cannot_replace_promoted_guest_line() {
266291 let (gate, delayed, resume) = barrier();
......@@ -268,28 +293,17 @@ fn handover_stale_met_cannot_replace_promoted_guest_line() {
268293 fixture.releases.push(resume.clone());
269294 let mut host = Hello::new("Old host route".into(), None).unwrap();
270295 host.serves = Some(fixture.guest.share);
271 let (_old, _) = fixture.open(&host, false);
296 let (_old, _, _) = fixture.open(&host, false);
272297 delayed.recv_timeout(OPENING).unwrap();
273298 host.name = "Promoted host route".into();
274 let (_new, _) = fixture.open(&host, true);
275 let overlap = match fixture.observed.recv_timeout(OPENING) {
276 Ok(()) => {
277 assert_eq!(
278 fixture.guest.host.lock().unwrap().as_ref().unwrap().0.name,
279 host.name
280 );
281 resume.send(()).unwrap();
282 fixture.heard();
283 true
284 }
285 Err(mpsc::RecvTimeoutError::Timeout) => {
286 resume.send(()).unwrap();
287 fixture.heard();
288 fixture.heard();
289 false
290 }
291 Err(error) => panic!("handover event channel closed: {error}"),
292 };
299 let (_new, _, _) = fixture.open(&host, true);
300 fixture.heard();
301 assert_eq!(
302 fixture.guest.host.lock().unwrap().as_ref().unwrap().0.name,
303 host.name
304 );
305 resume.send(()).unwrap();
306 fixture.heard();
293307 assert_eq!(
294308 fixture.shared.state.lock().unwrap().peers[&host.peer]
295309 .peer
......@@ -308,7 +322,6 @@ fn handover_stale_met_cannot_replace_promoted_guest_line() {
308322 .name
309323 .clone();
310324 drop(fixture);
311 println!("HANDOVER_MET_PROMOTION_OVERLAP={overlap}");
312325 assert_eq!(
313326 selected, host.name,
314327 "stale Met replaced the promoted guest Line"
......@@ -364,12 +377,12 @@ fn handover_old_left_preserves_promoted_host_and_pending_reply() {
364377 });
365378 let mut host = Hello::new("Old host route".into(), None).unwrap();
366379 host.serves = Some(fixture.guest.share);
367 let (old, _) = fixture.open(&host, false);
380 let (old, _, _) = fixture.open(&host, false);
368381 observed(&events, kind::HELLO);
369382 old.shutdown(Shutdown::Both).unwrap();
370383 removed.recv_timeout(OPENING).unwrap();
371384 host.name = "Promoted host route".into();
372 let (_new, _) = fixture.open(&host, true);
385 let (_new, _, _) = fixture.open(&host, true);
373386 observed(&events, kind::HELLO);
374387 let (answer, waiting) = crate::task::response::<Reply>();
375388 fixture.guest.pending.lock().unwrap().insert(73, answer);
......@@ -396,7 +409,7 @@ fn handover_selected_loss_clears_host_and_disconnects_pending_reply() {
396409 let mut fixture = Handover::new(None);
397410 let mut host = Hello::new("Selected host".into(), None).unwrap();
398411 host.serves = Some(fixture.guest.share);
399 let (remote, _) = fixture.open(&host, true);
412 let (remote, _, _) = fixture.open(&host, true);
400413 fixture.heard();
401414 let (answer, waiting) = crate::task::response::<Reply>();
402415 fixture.guest.pending.lock().unwrap().insert(74, answer);
......@@ -434,34 +447,17 @@ fn served() -> (tempfile::TempDir, Arc<Served>) {
434447#[test]
435448fn handover_copied_host_request_cannot_enter_promoted_queue() {
436449 let (_directory, served) = served();
437 let queues = Arc::new(Mutex::new(Vec::new()));
438450 let mut fixture = Handover::new(None);
439451 let serving = Arc::clone(&served);
440 let received = Arc::clone(&queues);
441452 let events = fixture.intercept(move |source, event| match event {
442 Event::Met(hello, line) => {
443 let (queue, waiting) = mpsc::sync_channel(4);
444 let retired = serving.guests.lock().unwrap().insert(
445 hello.peer,
446 Admitted {
447 hello: Arc::clone(hello),
448 line: (*line).clone(),
449 queue,
450 starts: BURST,
451 counted: Instant::now(),
452 held: Vec::new(),
453 },
454 );
455 drop(retired);
456 received.lock().unwrap().push(waiting);
457 }
453 Event::Met(hello, line) => assert!(serving.admit(source, hello, line)),
458454 Event::Frame { from, kind, body } => serving.queue(source, from, *kind, body),
459455 Event::Left(hello) => serving.forget(&hello.peer, Some((source, hello))),
460456 _ => {}
461457 });
462458 let mut host = Hello::new("Old route".into(), None).unwrap();
463459 host.serves = Some(fixture.guest.share);
464 let (mut old, mut send) = fixture.open(&host, false);
460 let (mut old, mut send, _) = fixture.open(&host, false);
465461 observed(&events, kind::HELLO);
466462 let (gate, copied, resume) = barrier();
467463 fixture.releases.push(resume.clone());
......@@ -471,6 +467,8 @@ fn handover_copied_host_request_cannot_enter_promoted_queue() {
471467 kind::PUT,
472468 &Request {
473469 id: 80,
470 handle: Some(1),
471 offset: Some(0),
474472 bytes: Some(vec![1, 2, 3]),
475473 ..Request::default()
476474 },
......@@ -478,31 +476,41 @@ fn handover_copied_host_request_cannot_enter_promoted_queue() {
478476 .unwrap();
479477 copied.recv_timeout(OPENING).unwrap();
480478 host.name = "Promoted route".into();
481 let (mut new, mut send) = fixture.open(&host, true);
479 let (mut new, mut send, mut receive) = fixture.open(&host, true);
482480 observed(&events, kind::HELLO);
483481 resume.send(()).unwrap();
484482 fixture.readers[0].0.recv_timeout(OPENING).unwrap();
485 let no_stale_request = matches!(
486 queues.lock().unwrap()[1].try_recv(),
487 Err(mpsc::TryRecvError::Empty)
488 );
489483 send.send(
490484 &mut new,
491485 kind::PUT,
492486 &Request {
493487 id: 81,
488 handle: Some(2),
489 offset: Some(0),
494490 bytes: Some(vec![4, 5, 6]),
495491 ..Request::default()
496492 },
497493 )
498494 .unwrap();
499 let (kind, body) = queues.lock().unwrap()[1].recv_timeout(OPENING).unwrap();
500 let accepted = kind == kind::PUT && minicbor::decode::<Request>(&body).unwrap().id == 81;
495 let reply = loop {
496 let (kind, body) = receive.receive(&mut new).unwrap();
497 if kind == kind::REPLY {
498 break minicbor::decode::<Reply>(&body).unwrap();
499 }
500 };
501 let puts = served
502 .puts
503 .lock()
504 .unwrap()
505 .keys()
506 .copied()
507 .collect::<Vec<_>>();
501508 served.guests.lock().unwrap().clear();
502509 drop(fixture);
503 assert!(
504 no_stale_request && accepted,
505 "retired request entered the promoted queue or current request was lost"
510 assert_eq!(
511 (reply.id, reply.failure, puts),
512 (81, None, vec![(host.peer, 2)]),
513 "retired request reached the promoted guest or current request was lost"
506514 );
507515}
508516
......@@ -514,25 +522,8 @@ fn handover_old_left_preserves_promoted_host_uploads_and_snapshots() {
514522 fixture.releases.push(resume.clone());
515523 let gate = Mutex::new(Some(gate));
516524 let serving = Arc::clone(&served);
517 let waiting = Arc::new(Mutex::new(Vec::new()));
518 let keep_queues = Arc::clone(&waiting);
519525 let events = fixture.intercept(move |source, event| match event {
520 Event::Met(hello, line) => {
521 let (queue, received) = mpsc::sync_channel(1);
522 let retired = serving.guests.lock().unwrap().insert(
523 hello.peer,
524 Admitted {
525 hello: Arc::clone(hello),
526 line: (*line).clone(),
527 queue,
528 starts: BURST,
529 counted: Instant::now(),
530 held: Vec::new(),
531 },
532 );
533 drop(retired);
534 keep_queues.lock().unwrap().push(received);
535 }
526 Event::Met(hello, line) => assert!(serving.admit(source, hello, line)),
536527 Event::Left(hello) => {
537528 if let Some(gate) = gate.lock().unwrap().take() {
538529 gate.wait().unwrap();
......@@ -543,12 +534,12 @@ fn handover_old_left_preserves_promoted_host_uploads_and_snapshots() {
543534 });
544535 let mut host = Hello::new("Old route".into(), None).unwrap();
545536 host.serves = Some(fixture.guest.share);
546 let (old, _) = fixture.open(&host, false);
537 let (old, _, _) = fixture.open(&host, false);
547538 observed(&events, kind::HELLO);
548539 old.shutdown(Shutdown::Both).unwrap();
549540 removed.recv_timeout(OPENING).unwrap();
550541 host.name = "Promoted route".into();
551 let (_new, _) = fixture.open(&host, true);
542 let (_new, _, _) = fixture.open(&host, true);
552543 observed(&events, kind::HELLO);
553544 served
554545 .puts
......@@ -588,18 +579,10 @@ fn handover_deferred_presence_cannot_replace_promoted_host_room() {
588579 let mut fixture = Handover::new(None);
589580 let mut host = Hello::new("Old host route".into(), None).unwrap();
590581 host.serves = Some(fixture.guest.share);
591 let (_old, _) = fixture.open(&host, false);
582 let (_old, _, _) = fixture.open(&host, false);
592583 fixture.heard();
593584 let from = Arc::clone(&fixture.guest.host.lock().unwrap().as_ref().unwrap().0);
594585 let secret = [61; 16];
595 let live = Live::start(
596 fixture.guest.me.clone(),
597 &Room::Notebook(secret),
598 None,
599 None,
600 |_| {},
601 )
602 .unwrap();
603586 let guest = Arc::clone(&fixture.guest);
604587 let source = Arc::clone(&fixture.shared);
605588 let here_owner = Arc::clone(&fixture.guest);
......@@ -608,14 +591,14 @@ fn handover_deferred_presence_cannot_replace_promoted_host_room() {
608591 let (finished, done) = mpsc::channel();
609592 let installer = thread::spawn(move || {
610593 entered.send(()).unwrap();
611 guest.install_presence(&source, &from, secret, live);
594 guest.join_presence(&source, &from, secret);
612595 let _ = finished.send(());
613596 true
614597 });
615598 fixture.readers.push((done, installer));
616599 started.recv_timeout(OPENING).unwrap();
617600 host.name = "Promoted host route".into();
618 let (_new, _) = fixture.open(&host, true);
601 let (_new, _, _) = fixture.open(&host, true);
619602 fixture.heard();
620603 drop(here);
621604 fixture.readers[1].0.recv_timeout(OPENING).unwrap();
......@@ -639,18 +622,10 @@ fn handover_presence_install_seeds_latest_here() {
639622 let mut fixture = Handover::new(None);
640623 let mut host = Hello::new("Current host".into(), None).unwrap();
641624 host.serves = Some(fixture.guest.share);
642 let (_remote, _) = fixture.open(&host, true);
625 let (_remote, _, _) = fixture.open(&host, true);
643626 fixture.heard();
644627 let from = Arc::clone(&fixture.guest.host.lock().unwrap().as_ref().unwrap().0);
645628 let secret = [62; 16];
646 let live = Live::start(
647 fixture.guest.me.clone(),
648 &Room::Notebook(secret),
649 None,
650 None,
651 |_| {},
652 )
653 .unwrap();
654629 let guest = Arc::clone(&fixture.guest);
655630 let source = Arc::clone(&fixture.shared);
656631 let mut here = fixture.guest.here.lock().unwrap();
......@@ -662,7 +637,7 @@ fn handover_presence_install_seeds_latest_here() {
662637 let (finished, done) = mpsc::channel();
663638 let installer = thread::spawn(move || {
664639 entered.send(()).unwrap();
665 guest.install_presence(&source, &from, secret, live);
640 guest.join_presence(&source, &from, secret);
666641 let _ = finished.send(());
667642 true
668643 });
......@@ -699,7 +674,7 @@ pub(super) fn promoted_routes<T>(
699674 shared.secret = room.secret();
700675 shared.room = room;
701676 let host = Hello::new("Pairing device".into(), None).unwrap();
702 let (_old, _) = fixture.open(&host, false);
677 let (_old, _, _) = fixture.open(&host, false);
703678 fixture.heard();
704679 let (old, old_line) = {
705680 let state = fixture.shared.state.lock().unwrap();
......@@ -707,7 +682,7 @@ pub(super) fn promoted_routes<T>(
707682 (Arc::clone(&link.peer.hello), link.line.clone())
708683 };
709684 before(&fixture.shared, &old, &old_line);
710 let (_new, _) = fixture.open(&host, true);
685 let (_new, _, _) = fixture.open(&host, true);
711686 fixture.heard();
712687 let (new, new_line) = {
713688 let state = fixture.shared.state.lock().unwrap();
......@@ -719,3 +694,147 @@ pub(super) fn promoted_routes<T>(
719694 drop(fixture);
720695 result
721696}
697
698#[test]
699fn handover_pair_pending_keeps_promoted_identity_and_rejects_retired_allow() {
700 let directory = tempfile::tempdir().unwrap();
701 let folder = directory.path().join("notebook");
702 std::fs::create_dir(&folder).unwrap();
703 let sharing = Sharing {
704 approve: true,
705 code: super::super::super::code::format(701, "ABCDEF").unwrap(),
706 ..Sharing::new("").unwrap()
707 };
708 let room = Room::share(&sharing.code, &sharing.password);
709 let host = Host::start(
710 crate::session::Notebook::open(&folder, directory.path().join("cache"))
711 .unwrap()
712 .into_storage(),
713 Hello::new("Notebook owner".into(), None).unwrap(),
714 sharing,
715 "Fixture",
716 None,
717 None,
718 || {},
719 |_| Ok(()),
720 )
721 .unwrap();
722 let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared);
723 let valid = promoted_routes(
724 room,
725 |source, old, old_line| {
726 (callback.events)(source, Event::Met(old, old_line));
727 },
728 |source, old, old_line, new, new_line| {
729 host.allow(&old.peer).unwrap();
730 let no_grant = host.sharing().members.is_empty();
731 (callback.events)(source, Event::Met(new, new_line));
732 (callback.events)(source, Event::Met(old, old_line));
733 (callback.events)(source, Event::Left(old));
734 let kept = host
735 .requests()
736 .first()
737 .is_some_and(|request| Arc::ptr_eq(request, new));
738 host.stop();
739 kept && no_grant
740 },
741 );
742 assert!(
743 valid,
744 "pairing admitted a retired route or retired Left removed the promoted request"
745 );
746}
747
748#[test]
749fn handover_pair_retired_met_cannot_persist_another_credential() {
750 let directory = tempfile::tempdir().unwrap();
751 let folder = directory.path().join("notebook");
752 std::fs::create_dir(&folder).unwrap();
753 let sharing = Sharing {
754 approve: false,
755 code: super::super::super::code::format(702, "ABCDEF").unwrap(),
756 ..Sharing::new("").unwrap()
757 };
758 let room = Room::share(&sharing.code, &sharing.password);
759 let kept = Arc::new(AtomicUsize::new(0));
760 let count = Arc::clone(&kept);
761 let host = Host::start(
762 crate::session::Notebook::open(&folder, directory.path().join("cache"))
763 .unwrap()
764 .into_storage(),
765 Hello::new("Notebook owner".into(), None).unwrap(),
766 sharing,
767 "Fixture",
768 None,
769 None,
770 || {},
771 move |_| {
772 count.fetch_add(1, Ordering::SeqCst);
773 Ok(())
774 },
775 )
776 .unwrap();
777 let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared);
778 let valid = promoted_routes(
779 room,
780 |_, _, _| {},
781 |source, old, old_line, new, new_line| {
782 (callback.events)(source, Event::Met(new, new_line));
783 let after_current = kept.load(Ordering::SeqCst);
784 (callback.events)(source, Event::Met(old, old_line));
785 let refused = kept.load(Ordering::SeqCst) == after_current
786 && after_current == 2
787 && host.sharing().members.len() == 1;
788 host.stop();
789 refused
790 },
791 );
792 assert!(valid, "retired pairing Met persisted another credential");
793}
794
795#[test]
796fn handover_retired_membership_rejects_deferred_access_met() {
797 let directory = tempfile::tempdir().unwrap();
798 let folder = directory.path().join("notebook");
799 std::fs::create_dir(&folder).unwrap();
800 let secret = [71; 16];
801 let sharing = Sharing {
802 code: super::super::super::code::format(703, "ABCDEF").unwrap(),
803 members: vec![Device {
804 secret,
805 name: "Retired device".into(),
806 device: None,
807 }],
808 ..Sharing::new("").unwrap()
809 };
810 let host = Host::start(
811 crate::session::Notebook::open(&folder, directory.path().join("cache"))
812 .unwrap()
813 .into_storage(),
814 Hello::new("Notebook owner".into(), None).unwrap(),
815 sharing,
816 "Fixture",
817 None,
818 None,
819 || {},
820 |_| Ok(()),
821 )
822 .unwrap();
823 let callback = Arc::clone(&host.members.access.lock().unwrap()[&secret].shared);
824 let refused = promoted_routes(
825 Room::Notebook(secret),
826 |_, _, _| {},
827 |source, _, _, current, line| {
828 host.remove(&secret).unwrap();
829 (callback.events)(source, Event::Met(current, line));
830 let refused = host.members.served.guests.lock().unwrap().is_empty()
831 && host.sharing().members.is_empty();
832 host.stop();
833 refused
834 },
835 );
836 assert!(
837 refused,
838 "deferred access callback readmitted a removed device"
839 );
840}
crates/notebook/src/live/share/membership.rs+31-188
......@@ -154,32 +154,27 @@ impl Host {
154154 Ok(())
155155 }
156156
157 /// Admits the device asking as `peer`, or drops its request if it has gone.
157158 pub fn allow(&self, peer: &[u8; 16]) -> io::Result<()> {
158 let request = {
159 let mut pending = self.members.pending.lock().unwrap();
160 if !pending.get(peer).is_some_and(|(hello, _, source)| {
161 source
162 .upgrade()
163 .is_some_and(|source| source.connected(hello))
164 }) {
165 return Ok(());
166 }
167 pending.remove(peer)
159 let request = self.members.pending.lock().unwrap().remove(peer);
160 let Some((hello, line, source)) = request else {
161 return Ok(());
168162 };
169 if let Some((hello, line, source)) = request {
170 if let Err(error) = self.members.grant(&hello, &line) {
171 let mut pending = self.members.pending.lock().unwrap();
172 if source
173 .upgrade()
174 .is_some_and(|source| source.connected(&hello))
175 && !pending.contains_key(peer)
176 {
177 pending.insert(*peer, (hello, line, source));
178 }
179 return Err(error);
163 let connected = || {
164 source
165 .upgrade()
166 .is_some_and(|source| source.connected(&hello))
167 };
168 if connected()
169 && let Err(error) = self.members.grant(&hello, &line)
170 {
171 let mut pending = self.members.pending.lock().unwrap();
172 if connected() && !pending.contains_key(peer) {
173 pending.insert(*peer, (hello, line, source));
180174 }
181 (self.members.events)();
175 return Err(error);
182176 }
177 (self.members.events)();
183178 Ok(())
184179 }
185180
......@@ -222,26 +217,24 @@ impl Host {
222217 let access = self.members.access.lock().unwrap();
223218 next.members
224219 .iter()
225 .filter_map(|device| {
226 let live = access.get(&device.secret)?;
227 let lines = live
228 .peers()
229 .into_iter()
220 .filter_map(|device| Some((device.secret, access.get(&device.secret)?)))
221 .flat_map(|(secret, live)| {
222 let welcome = self.members.welcome(&next, secret);
223 (live.peers().into_iter())
230224 .filter_map(|peer| live.line(&peer.hello.peer))
231 .collect::<Vec<_>>();
232 Some((self.members.welcome(&next, device.secret), lines))
225 .map(move |line| (welcome.clone(), line))
233226 })
234227 .collect::<Vec<_>>()
235228 };
229 // Browser send failures may reenter a removal that rotates the room again.
236230 #[cfg(target_arch = "wasm32")]
237231 drop(_change);
238 'welcomes: for (welcome, lines) in welcomes {
239 for line in lines {
240 if self.members.sharing.lock().unwrap().secret != welcome.room {
241 break 'welcomes;
242 }
243 let _ = line.send(kind::WELCOME, &welcome);
232 for (welcome, line) in welcomes {
233 #[cfg(target_arch = "wasm32")]
234 if self.members.sharing.lock().unwrap().secret != next.secret {
235 break;
244236 }
237 let _ = line.send(kind::WELCOME, &welcome);
245238 }
246239 (self.members.events)();
247240 Ok(())
......@@ -334,11 +327,7 @@ impl Members {
334327 Ok(Some(welcome)) => {
335328 let _ = line.send(kind::WELCOME, &welcome);
336329 }
337 Err(()) => {
338 #[cfg(not(target_arch = "wasm32"))]
339 drop(sharing);
340 line.hang_up("removed");
341 }
330 Err(()) => line.hang_up("removed"),
342331 Ok(None) => {}
343332 }
344333 }
......@@ -469,151 +458,5 @@ impl Members {
469458}
470459
471460#[cfg(all(test, not(target_arch = "wasm32")))]
472mod handover_tests {
473 use super::*;
474 use std::sync::atomic::AtomicUsize;
475
476 #[test]
477 fn handover_pair_pending_keeps_promoted_identity_and_rejects_retired_allow() {
478 let directory = tempfile::tempdir().unwrap();
479 let folder = directory.path().join("notebook");
480 std::fs::create_dir(&folder).unwrap();
481 let sharing = Sharing {
482 approve: true,
483 code: super::super::super::code::format(701, "ABCDEF").unwrap(),
484 ..Sharing::new("").unwrap()
485 };
486 let room = Room::share(&sharing.code, &sharing.password);
487 let host = Host::start(
488 crate::session::Notebook::open(&folder, directory.path().join("cache"))
489 .unwrap()
490 .into_storage(),
491 Hello::new("Notebook owner".into(), None).unwrap(),
492 sharing,
493 "Fixture",
494 None,
495 None,
496 || {},
497 |_| Ok(()),
498 )
499 .unwrap();
500 let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared);
501 let valid = super::super::handover_tests::promoted_routes(
502 room,
503 |source, old, old_line| {
504 (callback.events)(source, Event::Met(old, old_line));
505 },
506 |source, old, old_line, new, new_line| {
507 host.allow(&old.peer).unwrap();
508 let no_grant = host.sharing().members.is_empty();
509 (callback.events)(source, Event::Met(new, new_line));
510 (callback.events)(source, Event::Met(old, old_line));
511 (callback.events)(source, Event::Left(old));
512 let kept = host
513 .requests()
514 .first()
515 .is_some_and(|request| Arc::ptr_eq(request, new));
516 host.stop();
517 kept && no_grant
518 },
519 );
520 assert!(
521 valid,
522 "pairing admitted a retired route or retired Left removed the promoted request"
523 );
524 }
525
526 #[test]
527 fn handover_pair_retired_met_cannot_persist_another_credential() {
528 let directory = tempfile::tempdir().unwrap();
529 let folder = directory.path().join("notebook");
530 std::fs::create_dir(&folder).unwrap();
531 let sharing = Sharing {
532 approve: false,
533 code: super::super::super::code::format(702, "ABCDEF").unwrap(),
534 ..Sharing::new("").unwrap()
535 };
536 let room = Room::share(&sharing.code, &sharing.password);
537 let kept = Arc::new(AtomicUsize::new(0));
538 let count = Arc::clone(&kept);
539 let host = Host::start(
540 crate::session::Notebook::open(&folder, directory.path().join("cache"))
541 .unwrap()
542 .into_storage(),
543 Hello::new("Notebook owner".into(), None).unwrap(),
544 sharing,
545 "Fixture",
546 None,
547 None,
548 || {},
549 move |_| {
550 count.fetch_add(1, Ordering::SeqCst);
551 Ok(())
552 },
553 )
554 .unwrap();
555 let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared);
556 let valid = super::super::handover_tests::promoted_routes(
557 room,
558 |_, _, _| {},
559 |source, old, old_line, new, new_line| {
560 (callback.events)(source, Event::Met(new, new_line));
561 let after_current = kept.load(Ordering::SeqCst);
562 (callback.events)(source, Event::Met(old, old_line));
563 let refused = kept.load(Ordering::SeqCst) == after_current
564 && after_current == 2
565 && host.sharing().members.len() == 1;
566 host.stop();
567 refused
568 },
569 );
570 assert!(valid, "retired pairing Met persisted another credential");
571 }
572
573 #[test]
574 fn handover_retired_membership_rejects_deferred_access_met() {
575 let directory = tempfile::tempdir().unwrap();
576 let folder = directory.path().join("notebook");
577 std::fs::create_dir(&folder).unwrap();
578 let secret = [71; 16];
579 let sharing = Sharing {
580 code: super::super::super::code::format(703, "ABCDEF").unwrap(),
581 members: vec![Device {
582 secret,
583 name: "Retired device".into(),
584 device: None,
585 }],
586 ..Sharing::new("").unwrap()
587 };
588 let host = Host::start(
589 crate::session::Notebook::open(&folder, directory.path().join("cache"))
590 .unwrap()
591 .into_storage(),
592 Hello::new("Notebook owner".into(), None).unwrap(),
593 sharing,
594 "Fixture",
595 None,
596 None,
597 || {},
598 |_| Ok(()),
599 )
600 .unwrap();
601 let callback = Arc::clone(&host.members.access.lock().unwrap()[&secret].shared);
602 let refused = super::super::handover_tests::promoted_routes(
603 Room::Notebook(secret),
604 |_, _, _| {},
605 |source, _, _, current, line| {
606 host.remove(&secret).unwrap();
607 (callback.events)(source, Event::Met(current, line));
608 let refused = host.members.served.guests.lock().unwrap().is_empty()
609 && host.sharing().members.is_empty();
610 host.stop();
611 refused
612 },
613 );
614 assert!(
615 refused,
616 "deferred access callback readmitted a removed device"
617 );
618 }
619}
461#[path = "handover_tests.rs"]
462mod handover_tests;
crates/notebook/src/live/web.rs+26-28
......@@ -36,7 +36,9 @@ extern "C" {
3636 fn close(id: u32);
3737}
3838
39pub struct Live(Arc<Shared>);
39pub struct Live {
40 shared: Arc<Shared>,
41}
4042
4143struct Shared {
4244 me: Hello,
......@@ -209,35 +211,35 @@ impl Live {
209211 }
210212 }
211213 })?;
212 Ok(Self(shared))
214 Ok(Self { shared })
213215 }
214216
215217 pub fn code(&self) -> Option<String> {
216 match &self.0.room {
218 match &self.shared.room {
217219 Room::Code { code, .. } => Some(code.clone()),
218220 _ => None,
219221 }
220222 }
221223 pub fn relayed(&self) -> Relayed {
222 self.0.state.lock().unwrap().relayed.clone()
224 self.shared.state.lock().unwrap().relayed.clone()
223225 }
224226 pub fn failed(&self) -> u32 {
225 self.0.state.lock().unwrap().failed
227 self.shared.state.lock().unwrap().failed
226228 }
227229 pub fn other_version(&self) -> Option<u16> {
228 self.0.state.lock().unwrap().outdated
230 self.shared.state.lock().unwrap().outdated
229231 }
230232 pub fn burned(&self) -> bool {
231 self.0.state.lock().unwrap().burned
233 self.shared.state.lock().unwrap().burned
232234 }
233235 pub fn sender(&self) -> Sender {
234 Sender(Arc::downgrade(&self.0))
236 Sender(Arc::downgrade(&self.shared))
235237 }
236238 pub fn peers(&self) -> Vec<Peer> {
237 self.0.peers()
239 self.shared.peers()
238240 }
239241 pub fn line(&self, peer: &[u8; 16]) -> Option<Line> {
240 self.0
242 self.shared
241243 .state
242244 .lock()
243245 .unwrap()
......@@ -247,7 +249,7 @@ impl Live {
247249 .map(|s| s.line.clone())
248250 }
249251 pub fn set_presence(&self, presence: Presence) {
250 let mut state = self.0.state.lock().unwrap();
252 let mut state = self.shared.state.lock().unwrap();
251253 if state.presence == presence {
252254 return;
253255 }
......@@ -256,7 +258,7 @@ impl Live {
256258 return;
257259 }
258260 drop(state);
259 let weak = Arc::downgrade(&self.0);
261 let weak = Arc::downgrade(&self.shared);
260262 let _ = crate::task::spawn("live presence", move || async move {
261263 let (_, wait) = crate::task::channel();
262264 crate::task::wait(&wait, Some(Duration::from_millis(100))).await;
......@@ -272,7 +274,7 @@ impl Live {
272274 }
273275 pub fn leave(self, reason: &str) {
274276 let lines: Vec<_> = self
275 .0
277 .shared
276278 .state
277279 .lock()
278280 .unwrap()
......@@ -288,8 +290,8 @@ impl Live {
288290
289291impl Drop for Live {
290292 fn drop(&mut self) {
291 self.0.stopped.store(true, Ordering::Release);
292 let socket = self.0.state.lock().unwrap().socket;
293 self.shared.stopped.store(true, Ordering::Release);
294 let socket = self.shared.state.lock().unwrap().socket;
293295 close(socket);
294296 }
295297}
......@@ -310,22 +312,18 @@ impl Shared {
310312 }
311313
312314 fn gone(&self, hello: &Arc<Hello>) -> bool {
313 !self.state.lock().unwrap().streams.values().any(|stream| {
314 stream
315 .peer
316 .as_ref()
317 .is_some_and(|peer| Arc::ptr_eq(&peer.hello, hello))
318 })
315 let state = self.state.lock().unwrap();
316 !(state.members.values().map(|member| &member.peer))
317 .chain(state.streams.values().map(|stream| &stream.peer))
318 .any(|peer| {
319 peer.as_ref()
320 .is_some_and(|peer| Arc::ptr_eq(&peer.hello, hello))
321 })
319322 }
320323
324 /// Whether `hello` is how a peer is in the room now: its stream, or its place in the group.
321325 fn connected(&self, hello: &Arc<Hello>) -> bool {
322 !self.stopped.load(Ordering::Acquire)
323 && self.state.lock().unwrap().streams.values().any(|stream| {
324 stream
325 .peer
326 .as_ref()
327 .is_some_and(|peer| Arc::ptr_eq(&peer.hello, hello))
328 })
326 !self.stopped.load(Ordering::Acquire) && !self.gone(hello)
329327 }
330328
331329 fn connect(self: &Arc<Self>) -> io::Result<()> {