diff --git a/crates/notebook/src/live.rs b/crates/notebook/src/live.rs index acd6687f01d6159b33e77cd431638e78bb0c0495..f60240fc848c4eca71c7cece8a431146efa5d04d 100644 --- a/crates/notebook/src/live.rs +++ b/crates/notebook/src/live.rs @@ -455,15 +455,19 @@ impl Shared { !self.state.lock().unwrap().peers.contains_key(&hello.peer) } + /// Whether `hello` is how a peer is in the room now: its link, or its place in the group. fn connected(&self, hello: &Arc) -> bool { + let current = |peer: &Peer| Arc::ptr_eq(&peer.hello, hello); + let state = self.state.lock().unwrap(); !self.stopped.load(Ordering::Acquire) - && self - .state - .lock() - .unwrap() + && (state .peers .get(&hello.peer) - .is_some_and(|link| Arc::ptr_eq(&link.peer.hello, hello)) + .is_some_and(|link| current(&link.peer)) + || state.group().is_some_and(|group| { + (group.members.lock().unwrap().values()) + .any(|member| member.peer.as_ref().is_some_and(current)) + })) } /// The peers in the room: those met directly, and the rest as the relay's group or a diff --git a/crates/notebook/src/live/share.rs b/crates/notebook/src/live/share.rs index cfbd6dbf93113a9dcf1ea09124ea02b48bc75815..d0fd3c305336cd2eb48a06fa597342f95be619aa 100644 --- a/crates/notebook/src/live/share.rs +++ b/crates/notebook/src/live/share.rs @@ -29,9 +29,6 @@ use web_time::Instant; #[path = "share/batch.rs"] mod batch; -#[cfg(all(test, not(target_arch = "wasm32")))] -#[path = "share/handover_tests.rs"] -mod handover_tests; #[path = "share/membership.rs"] mod membership; #[cfg(target_arch = "wasm32")] @@ -1643,6 +1640,21 @@ impl Inner { reports.lost(); } } + // A reply only reaches its waiter, so one read off a retired route still counts. + Event::Frame { + from, + kind: kind::REPLY, + body, + } => { + let serves = (self.host.lock().unwrap().as_ref()) + .is_some_and(|(host, _)| host.peer == from.peer); + if serves && let Ok(reply) = minicbor::decode::(body) { + let waiting = self.pending.lock().unwrap().remove(&reply.id); + if let Some(waiting) = waiting { + let _ = waiting.send(reply); + } + } + } Event::Frame { from, kind, body } => { let host = self.host.lock().unwrap(); if !host @@ -1654,20 +1666,14 @@ impl Inner { } match kind { kind::WELCOME => { - if let Ok(welcome) = minicbor::decode::(body) { + if let Ok(welcome) = minicbor::decode::(body) + && !(self.presence.lock().unwrap().as_ref()) + .is_some_and(|(held, _)| *held == welcome.room) + { drop(host); self.join_presence(source, from, welcome.room); } } - kind::REPLY => { - if let Ok(reply) = minicbor::decode::(body) { - let waiting = self.pending.lock().unwrap().remove(&reply.id); - drop(host); - if let Some(waiting) = waiting { - let _ = waiting.send(reply); - } - } - } kind::DELTA => { if let Ok(delta) = minicbor::decode::(body) { self.apply(&delta); @@ -1708,22 +1714,6 @@ impl Inner { from: &Arc, secret: [u8; 16], ) { - { - let host = self.host.lock().unwrap(); - if !host - .as_ref() - .is_some_and(|(admitted, _)| Arc::ptr_eq(admitted, from)) - || !source.connected(from) - || self - .presence - .lock() - .unwrap() - .as_ref() - .is_some_and(|(held, _)| *held == secret) - { - return; - } - } let inner = Arc::downgrade(self); let connected = Mutex::new(None); let live = Live::start_with( @@ -1741,11 +1731,7 @@ impl Inner { let Some((_, live)) = presence.as_ref() else { return; }; - #[cfg(not(target_arch = "wasm32"))] - let selected = Arc::ptr_eq(&live.shared, source); - #[cfg(target_arch = "wasm32")] - let selected = Arc::ptr_eq(&live.0, source); - if !selected { + if !Arc::ptr_eq(&live.shared, source) { return; } match event { @@ -1776,10 +1762,7 @@ impl Inner { if host .as_ref() .is_some_and(|(host, _)| host.peer == from.peer) - && source - .peers() - .iter() - .any(|peer| Arc::ptr_eq(&peer.hello, from)) => + && source.connected(from) => { match kind { kind::DELTA => { @@ -1813,25 +1796,13 @@ impl Inner { (inner.events)(); }, ); - match live { - Ok(live) => self.install_presence(source, from, secret, live), - Err(error) => eprintln!("Live Share: could not join presence: {error}"), - } - } - - fn install_presence( - self: &Arc, - source: &Arc, - from: &Arc, - secret: [u8; 16], - live: Live, - ) { + let live = match live { + Ok(live) => live, + Err(error) => return eprintln!("Live Share: could not join presence: {error}"), + }; let here = self.here.lock().unwrap(); live.set_presence(here.clone()); - #[cfg(not(target_arch = "wasm32"))] let refresh = Arc::clone(&live.shared); - #[cfg(target_arch = "wasm32")] - let refresh = Arc::clone(&live.0); let mut candidate = Some(live); let (retired, paths, reports) = { let host = self.host.lock().unwrap(); diff --git a/crates/notebook/src/live/share/handover_tests.rs b/crates/notebook/src/live/share/handover_tests.rs index 564dc26ff817a4108a82b23ce37640ae2a02c8e2..feb9c7813e796dd970ff9dd16c1939f6910b2ef9 100644 --- a/crates/notebook/src/live/share/handover_tests.rs +++ b/crates/notebook/src/live/share/handover_tests.rs @@ -1,8 +1,9 @@ -use super::super::{GONE, OPENING, Pipe, Shared, State}; +use super::super::super::{GONE, OPENING, Pipe, Shared, State}; use super::*; use std::{ io::{Read, Write}, net::{Shutdown, TcpListener, TcpStream}, + sync::atomic::AtomicUsize, }; struct Barrier { @@ -144,7 +145,7 @@ impl Handover { } } - fn open(&mut self, host: &Hello, direct: bool) -> (TcpStream, wire::Sealer) { + fn open(&mut self, host: &Hello, direct: bool) -> (TcpStream, wire::Sealer, wire::Sealer) { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let mut remote = TcpStream::connect_timeout(&listener.local_addr().unwrap(), OPENING).unwrap(); @@ -180,7 +181,7 @@ impl Handover { let (message, body) = receive.receive(&mut remote).unwrap(); assert_eq!(message, kind::HELLO); assert_eq!(minicbor::decode::(&body).unwrap(), self.shared.me); - (remote, send) + (remote, send, receive) } fn heard(&self) { @@ -198,25 +199,17 @@ impl Drop for Handover { pipe.shutdown(); } self.shared.state.lock().unwrap().peers.clear(); - let deadline = std::time::Instant::now() + OPENING; - for (_, reader) in &self.readers { - while !reader.is_finished() && std::time::Instant::now() < deadline { - thread::yield_now(); - } - } - let readers_done = self.readers.iter().all(|(_, reader)| reader.is_finished()); + let readers_done = self + .readers + .iter() + .all(|(done, _)| done.recv_timeout(OPENING) != Err(mpsc::RecvTimeoutError::Timeout)); self.guest.host.lock().unwrap().take(); let count = self.pipes.len(); self.pipes.clear(); - let retired = (0..count).all(|_| self.retired.1.recv_timeout(OPENING).is_ok()); - let mut joined = readers_done && retired; - if readers_done { - for (_, reader) in self.readers.drain(..) { - joined &= reader.join().is_ok(); - } - } + let joined = readers_done + && (0..count).all(|_| self.retired.1.recv_timeout(OPENING).is_ok()) + && (self.readers.drain(..)).all(|(_, reader)| reader.join().is_ok()); if !thread::panicking() { - println!("HANDOVER_READERS_JOINED_AND_PIPE_OWNERS_RETIRED={joined};PIPES={count}"); assert!( joined, "handover reader/writer owners did not retire and join" @@ -230,7 +223,7 @@ fn handover_stale_frame_cannot_end_promoted_guest() { let mut fixture = Handover::new(None); let mut host = Hello::new("Old host route".into(), None).unwrap(); host.serves = Some(fixture.guest.share); - let (mut old, mut send) = fixture.open(&host, false); + let (mut old, mut send, _) = fixture.open(&host, false); fixture.heard(); let (gate, copied, resume) = barrier(); fixture.releases.push(resume.clone()); @@ -245,7 +238,7 @@ fn handover_stale_frame_cannot_end_promoted_guest() { .unwrap(); copied.recv_timeout(OPENING).unwrap(); host.name = "Promoted host route".into(); - let (_new, _) = fixture.open(&host, true); + let (_new, _, _) = fixture.open(&host, true); fixture.heard(); assert!( fixture.shared.state.lock().unwrap().peers[&host.peer] @@ -257,10 +250,42 @@ fn handover_stale_frame_cannot_end_promoted_guest() { fixture.readers[0].0.recv_timeout(OPENING).unwrap(); let ended = *fixture.guest.ended.lock().unwrap(); drop(fixture); - println!("HANDOVER_STALE_FRAME_ENDED={ended:?}"); assert_eq!(ended, None, "stale frame ended the promoted guest"); } +#[test] +fn handover_reply_from_retired_route_reaches_its_waiter() { + let mut fixture = Handover::new(None); + let mut host = Hello::new("Old host route".into(), None).unwrap(); + host.serves = Some(fixture.guest.share); + let (mut old, mut send, _) = fixture.open(&host, false); + fixture.heard(); + let (answer, waiting) = crate::task::response::(); + fixture.guest.pending.lock().unwrap().insert(75, answer); + let (gate, copied, resume) = barrier(); + fixture.releases.push(resume.clone()); + *fixture.pipes[0].body.lock().unwrap() = Some(gate); + let reply = Reply { + id: 75, + length: Some(12), + ..Reply::default() + }; + send.send(&mut old, kind::REPLY, &reply).unwrap(); + copied.recv_timeout(OPENING).unwrap(); + host.name = "Promoted host route".into(); + let (_new, _, _) = fixture.open(&host, true); + fixture.heard(); + resume.send(()).unwrap(); + fixture.readers[0].0.recv_timeout(OPENING).unwrap(); + let answered = waiting.try_recv().ok(); + drop(fixture); + assert_eq!( + answered, + Some(reply), + "a reply read off a retired route missed its waiter" + ); +} + #[test] fn handover_stale_met_cannot_replace_promoted_guest_line() { let (gate, delayed, resume) = barrier(); @@ -268,28 +293,17 @@ fn handover_stale_met_cannot_replace_promoted_guest_line() { fixture.releases.push(resume.clone()); let mut host = Hello::new("Old host route".into(), None).unwrap(); host.serves = Some(fixture.guest.share); - let (_old, _) = fixture.open(&host, false); + let (_old, _, _) = fixture.open(&host, false); delayed.recv_timeout(OPENING).unwrap(); host.name = "Promoted host route".into(); - let (_new, _) = fixture.open(&host, true); - let overlap = match fixture.observed.recv_timeout(OPENING) { - Ok(()) => { - assert_eq!( - fixture.guest.host.lock().unwrap().as_ref().unwrap().0.name, - host.name - ); - resume.send(()).unwrap(); - fixture.heard(); - true - } - Err(mpsc::RecvTimeoutError::Timeout) => { - resume.send(()).unwrap(); - fixture.heard(); - fixture.heard(); - false - } - Err(error) => panic!("handover event channel closed: {error}"), - }; + let (_new, _, _) = fixture.open(&host, true); + fixture.heard(); + assert_eq!( + fixture.guest.host.lock().unwrap().as_ref().unwrap().0.name, + host.name + ); + resume.send(()).unwrap(); + fixture.heard(); assert_eq!( fixture.shared.state.lock().unwrap().peers[&host.peer] .peer @@ -308,7 +322,6 @@ fn handover_stale_met_cannot_replace_promoted_guest_line() { .name .clone(); drop(fixture); - println!("HANDOVER_MET_PROMOTION_OVERLAP={overlap}"); assert_eq!( selected, host.name, "stale Met replaced the promoted guest Line" @@ -364,12 +377,12 @@ fn handover_old_left_preserves_promoted_host_and_pending_reply() { }); let mut host = Hello::new("Old host route".into(), None).unwrap(); host.serves = Some(fixture.guest.share); - let (old, _) = fixture.open(&host, false); + let (old, _, _) = fixture.open(&host, false); observed(&events, kind::HELLO); old.shutdown(Shutdown::Both).unwrap(); removed.recv_timeout(OPENING).unwrap(); host.name = "Promoted host route".into(); - let (_new, _) = fixture.open(&host, true); + let (_new, _, _) = fixture.open(&host, true); observed(&events, kind::HELLO); let (answer, waiting) = crate::task::response::(); fixture.guest.pending.lock().unwrap().insert(73, answer); @@ -396,7 +409,7 @@ fn handover_selected_loss_clears_host_and_disconnects_pending_reply() { let mut fixture = Handover::new(None); let mut host = Hello::new("Selected host".into(), None).unwrap(); host.serves = Some(fixture.guest.share); - let (remote, _) = fixture.open(&host, true); + let (remote, _, _) = fixture.open(&host, true); fixture.heard(); let (answer, waiting) = crate::task::response::(); fixture.guest.pending.lock().unwrap().insert(74, answer); @@ -434,34 +447,17 @@ fn served() -> (tempfile::TempDir, Arc) { #[test] fn handover_copied_host_request_cannot_enter_promoted_queue() { let (_directory, served) = served(); - let queues = Arc::new(Mutex::new(Vec::new())); let mut fixture = Handover::new(None); let serving = Arc::clone(&served); - let received = Arc::clone(&queues); let events = fixture.intercept(move |source, event| match event { - Event::Met(hello, line) => { - let (queue, waiting) = mpsc::sync_channel(4); - let retired = serving.guests.lock().unwrap().insert( - hello.peer, - Admitted { - hello: Arc::clone(hello), - line: (*line).clone(), - queue, - starts: BURST, - counted: Instant::now(), - held: Vec::new(), - }, - ); - drop(retired); - received.lock().unwrap().push(waiting); - } + Event::Met(hello, line) => assert!(serving.admit(source, hello, line)), Event::Frame { from, kind, body } => serving.queue(source, from, *kind, body), Event::Left(hello) => serving.forget(&hello.peer, Some((source, hello))), _ => {} }); let mut host = Hello::new("Old route".into(), None).unwrap(); host.serves = Some(fixture.guest.share); - let (mut old, mut send) = fixture.open(&host, false); + let (mut old, mut send, _) = fixture.open(&host, false); observed(&events, kind::HELLO); let (gate, copied, resume) = barrier(); fixture.releases.push(resume.clone()); @@ -471,6 +467,8 @@ fn handover_copied_host_request_cannot_enter_promoted_queue() { kind::PUT, &Request { id: 80, + handle: Some(1), + offset: Some(0), bytes: Some(vec![1, 2, 3]), ..Request::default() }, @@ -478,31 +476,41 @@ fn handover_copied_host_request_cannot_enter_promoted_queue() { .unwrap(); copied.recv_timeout(OPENING).unwrap(); host.name = "Promoted route".into(); - let (mut new, mut send) = fixture.open(&host, true); + let (mut new, mut send, mut receive) = fixture.open(&host, true); observed(&events, kind::HELLO); resume.send(()).unwrap(); fixture.readers[0].0.recv_timeout(OPENING).unwrap(); - let no_stale_request = matches!( - queues.lock().unwrap()[1].try_recv(), - Err(mpsc::TryRecvError::Empty) - ); send.send( &mut new, kind::PUT, &Request { id: 81, + handle: Some(2), + offset: Some(0), bytes: Some(vec![4, 5, 6]), ..Request::default() }, ) .unwrap(); - let (kind, body) = queues.lock().unwrap()[1].recv_timeout(OPENING).unwrap(); - let accepted = kind == kind::PUT && minicbor::decode::(&body).unwrap().id == 81; + let reply = loop { + let (kind, body) = receive.receive(&mut new).unwrap(); + if kind == kind::REPLY { + break minicbor::decode::(&body).unwrap(); + } + }; + let puts = served + .puts + .lock() + .unwrap() + .keys() + .copied() + .collect::>(); served.guests.lock().unwrap().clear(); drop(fixture); - assert!( - no_stale_request && accepted, - "retired request entered the promoted queue or current request was lost" + assert_eq!( + (reply.id, reply.failure, puts), + (81, None, vec![(host.peer, 2)]), + "retired request reached the promoted guest or current request was lost" ); } @@ -514,25 +522,8 @@ fn handover_old_left_preserves_promoted_host_uploads_and_snapshots() { fixture.releases.push(resume.clone()); let gate = Mutex::new(Some(gate)); let serving = Arc::clone(&served); - let waiting = Arc::new(Mutex::new(Vec::new())); - let keep_queues = Arc::clone(&waiting); let events = fixture.intercept(move |source, event| match event { - Event::Met(hello, line) => { - let (queue, received) = mpsc::sync_channel(1); - let retired = serving.guests.lock().unwrap().insert( - hello.peer, - Admitted { - hello: Arc::clone(hello), - line: (*line).clone(), - queue, - starts: BURST, - counted: Instant::now(), - held: Vec::new(), - }, - ); - drop(retired); - keep_queues.lock().unwrap().push(received); - } + Event::Met(hello, line) => assert!(serving.admit(source, hello, line)), Event::Left(hello) => { if let Some(gate) = gate.lock().unwrap().take() { gate.wait().unwrap(); @@ -543,12 +534,12 @@ fn handover_old_left_preserves_promoted_host_uploads_and_snapshots() { }); let mut host = Hello::new("Old route".into(), None).unwrap(); host.serves = Some(fixture.guest.share); - let (old, _) = fixture.open(&host, false); + let (old, _, _) = fixture.open(&host, false); observed(&events, kind::HELLO); old.shutdown(Shutdown::Both).unwrap(); removed.recv_timeout(OPENING).unwrap(); host.name = "Promoted route".into(); - let (_new, _) = fixture.open(&host, true); + let (_new, _, _) = fixture.open(&host, true); observed(&events, kind::HELLO); served .puts @@ -588,18 +579,10 @@ fn handover_deferred_presence_cannot_replace_promoted_host_room() { let mut fixture = Handover::new(None); let mut host = Hello::new("Old host route".into(), None).unwrap(); host.serves = Some(fixture.guest.share); - let (_old, _) = fixture.open(&host, false); + let (_old, _, _) = fixture.open(&host, false); fixture.heard(); let from = Arc::clone(&fixture.guest.host.lock().unwrap().as_ref().unwrap().0); let secret = [61; 16]; - let live = Live::start( - fixture.guest.me.clone(), - &Room::Notebook(secret), - None, - None, - |_| {}, - ) - .unwrap(); let guest = Arc::clone(&fixture.guest); let source = Arc::clone(&fixture.shared); let here_owner = Arc::clone(&fixture.guest); @@ -608,14 +591,14 @@ fn handover_deferred_presence_cannot_replace_promoted_host_room() { let (finished, done) = mpsc::channel(); let installer = thread::spawn(move || { entered.send(()).unwrap(); - guest.install_presence(&source, &from, secret, live); + guest.join_presence(&source, &from, secret); let _ = finished.send(()); true }); fixture.readers.push((done, installer)); started.recv_timeout(OPENING).unwrap(); host.name = "Promoted host route".into(); - let (_new, _) = fixture.open(&host, true); + let (_new, _, _) = fixture.open(&host, true); fixture.heard(); drop(here); fixture.readers[1].0.recv_timeout(OPENING).unwrap(); @@ -639,18 +622,10 @@ fn handover_presence_install_seeds_latest_here() { let mut fixture = Handover::new(None); let mut host = Hello::new("Current host".into(), None).unwrap(); host.serves = Some(fixture.guest.share); - let (_remote, _) = fixture.open(&host, true); + let (_remote, _, _) = fixture.open(&host, true); fixture.heard(); let from = Arc::clone(&fixture.guest.host.lock().unwrap().as_ref().unwrap().0); let secret = [62; 16]; - let live = Live::start( - fixture.guest.me.clone(), - &Room::Notebook(secret), - None, - None, - |_| {}, - ) - .unwrap(); let guest = Arc::clone(&fixture.guest); let source = Arc::clone(&fixture.shared); let mut here = fixture.guest.here.lock().unwrap(); @@ -662,7 +637,7 @@ fn handover_presence_install_seeds_latest_here() { let (finished, done) = mpsc::channel(); let installer = thread::spawn(move || { entered.send(()).unwrap(); - guest.install_presence(&source, &from, secret, live); + guest.join_presence(&source, &from, secret); let _ = finished.send(()); true }); @@ -699,7 +674,7 @@ pub(super) fn promoted_routes( shared.secret = room.secret(); shared.room = room; let host = Hello::new("Pairing device".into(), None).unwrap(); - let (_old, _) = fixture.open(&host, false); + let (_old, _, _) = fixture.open(&host, false); fixture.heard(); let (old, old_line) = { let state = fixture.shared.state.lock().unwrap(); @@ -707,7 +682,7 @@ pub(super) fn promoted_routes( (Arc::clone(&link.peer.hello), link.line.clone()) }; before(&fixture.shared, &old, &old_line); - let (_new, _) = fixture.open(&host, true); + let (_new, _, _) = fixture.open(&host, true); fixture.heard(); let (new, new_line) = { let state = fixture.shared.state.lock().unwrap(); @@ -719,3 +694,147 @@ pub(super) fn promoted_routes( drop(fixture); result } + +#[test] +fn handover_pair_pending_keeps_promoted_identity_and_rejects_retired_allow() { + let directory = tempfile::tempdir().unwrap(); + let folder = directory.path().join("notebook"); + std::fs::create_dir(&folder).unwrap(); + let sharing = Sharing { + approve: true, + code: super::super::super::code::format(701, "ABCDEF").unwrap(), + ..Sharing::new("").unwrap() + }; + let room = Room::share(&sharing.code, &sharing.password); + let host = Host::start( + crate::session::Notebook::open(&folder, directory.path().join("cache")) + .unwrap() + .into_storage(), + Hello::new("Notebook owner".into(), None).unwrap(), + sharing, + "Fixture", + None, + None, + || {}, + |_| Ok(()), + ) + .unwrap(); + let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared); + let valid = promoted_routes( + room, + |source, old, old_line| { + (callback.events)(source, Event::Met(old, old_line)); + }, + |source, old, old_line, new, new_line| { + host.allow(&old.peer).unwrap(); + let no_grant = host.sharing().members.is_empty(); + (callback.events)(source, Event::Met(new, new_line)); + (callback.events)(source, Event::Met(old, old_line)); + (callback.events)(source, Event::Left(old)); + let kept = host + .requests() + .first() + .is_some_and(|request| Arc::ptr_eq(request, new)); + host.stop(); + kept && no_grant + }, + ); + assert!( + valid, + "pairing admitted a retired route or retired Left removed the promoted request" + ); +} + +#[test] +fn handover_pair_retired_met_cannot_persist_another_credential() { + let directory = tempfile::tempdir().unwrap(); + let folder = directory.path().join("notebook"); + std::fs::create_dir(&folder).unwrap(); + let sharing = Sharing { + approve: false, + code: super::super::super::code::format(702, "ABCDEF").unwrap(), + ..Sharing::new("").unwrap() + }; + let room = Room::share(&sharing.code, &sharing.password); + let kept = Arc::new(AtomicUsize::new(0)); + let count = Arc::clone(&kept); + let host = Host::start( + crate::session::Notebook::open(&folder, directory.path().join("cache")) + .unwrap() + .into_storage(), + Hello::new("Notebook owner".into(), None).unwrap(), + sharing, + "Fixture", + None, + None, + || {}, + move |_| { + count.fetch_add(1, Ordering::SeqCst); + Ok(()) + }, + ) + .unwrap(); + let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared); + let valid = promoted_routes( + room, + |_, _, _| {}, + |source, old, old_line, new, new_line| { + (callback.events)(source, Event::Met(new, new_line)); + let after_current = kept.load(Ordering::SeqCst); + (callback.events)(source, Event::Met(old, old_line)); + let refused = kept.load(Ordering::SeqCst) == after_current + && after_current == 2 + && host.sharing().members.len() == 1; + host.stop(); + refused + }, + ); + assert!(valid, "retired pairing Met persisted another credential"); +} + +#[test] +fn handover_retired_membership_rejects_deferred_access_met() { + let directory = tempfile::tempdir().unwrap(); + let folder = directory.path().join("notebook"); + std::fs::create_dir(&folder).unwrap(); + let secret = [71; 16]; + let sharing = Sharing { + code: super::super::super::code::format(703, "ABCDEF").unwrap(), + members: vec![Device { + secret, + name: "Retired device".into(), + device: None, + }], + ..Sharing::new("").unwrap() + }; + let host = Host::start( + crate::session::Notebook::open(&folder, directory.path().join("cache")) + .unwrap() + .into_storage(), + Hello::new("Notebook owner".into(), None).unwrap(), + sharing, + "Fixture", + None, + None, + || {}, + |_| Ok(()), + ) + .unwrap(); + let callback = Arc::clone(&host.members.access.lock().unwrap()[&secret].shared); + let refused = promoted_routes( + Room::Notebook(secret), + |_, _, _| {}, + |source, _, _, current, line| { + host.remove(&secret).unwrap(); + (callback.events)(source, Event::Met(current, line)); + let refused = host.members.served.guests.lock().unwrap().is_empty() + && host.sharing().members.is_empty(); + host.stop(); + refused + }, + ); + assert!( + refused, + "deferred access callback readmitted a removed device" + ); +} diff --git a/crates/notebook/src/live/share/membership.rs b/crates/notebook/src/live/share/membership.rs index 8b21f1fc64358601653b55680f7d45e1fa6d9ae6..3254d7ab3c7ba828e61a33eb1512a2f9b7b4a041 100644 --- a/crates/notebook/src/live/share/membership.rs +++ b/crates/notebook/src/live/share/membership.rs @@ -154,32 +154,27 @@ impl Host { Ok(()) } + /// Admits the device asking as `peer`, or drops its request if it has gone. pub fn allow(&self, peer: &[u8; 16]) -> io::Result<()> { - let request = { + let request = self.members.pending.lock().unwrap().remove(peer); + let Some((hello, line, source)) = request else { + return Ok(()); + }; + let connected = || { + source + .upgrade() + .is_some_and(|source| source.connected(&hello)) + }; + if connected() + && let Err(error) = self.members.grant(&hello, &line) + { let mut pending = self.members.pending.lock().unwrap(); - if !pending.get(peer).is_some_and(|(hello, _, source)| { - source - .upgrade() - .is_some_and(|source| source.connected(hello)) - }) { - return Ok(()); + if connected() && !pending.contains_key(peer) { + pending.insert(*peer, (hello, line, source)); } - pending.remove(peer) - }; - if let Some((hello, line, source)) = request { - if let Err(error) = self.members.grant(&hello, &line) { - let mut pending = self.members.pending.lock().unwrap(); - if source - .upgrade() - .is_some_and(|source| source.connected(&hello)) - && !pending.contains_key(peer) - { - pending.insert(*peer, (hello, line, source)); - } - return Err(error); - } - (self.members.events)(); + return Err(error); } + (self.members.events)(); Ok(()) } @@ -222,26 +217,24 @@ impl Host { let access = self.members.access.lock().unwrap(); next.members .iter() - .filter_map(|device| { - let live = access.get(&device.secret)?; - let lines = live - .peers() - .into_iter() + .filter_map(|device| Some((device.secret, access.get(&device.secret)?))) + .flat_map(|(secret, live)| { + let welcome = self.members.welcome(&next, secret); + (live.peers().into_iter()) .filter_map(|peer| live.line(&peer.hello.peer)) - .collect::>(); - Some((self.members.welcome(&next, device.secret), lines)) + .map(move |line| (welcome.clone(), line)) }) .collect::>() }; + // Browser send failures may reenter a removal that rotates the room again. #[cfg(target_arch = "wasm32")] drop(_change); - 'welcomes: for (welcome, lines) in welcomes { - for line in lines { - if self.members.sharing.lock().unwrap().secret != welcome.room { - break 'welcomes; - } - let _ = line.send(kind::WELCOME, &welcome); + for (welcome, line) in welcomes { + #[cfg(target_arch = "wasm32")] + if self.members.sharing.lock().unwrap().secret != next.secret { + break; } + let _ = line.send(kind::WELCOME, &welcome); } (self.members.events)(); Ok(()) @@ -334,11 +327,7 @@ impl Members { Ok(Some(welcome)) => { let _ = line.send(kind::WELCOME, &welcome); } - Err(()) => { - #[cfg(not(target_arch = "wasm32"))] - drop(sharing); - line.hang_up("removed"); - } + Err(()) => line.hang_up("removed"), Ok(None) => {} } } @@ -469,151 +458,5 @@ impl Members { } #[cfg(all(test, not(target_arch = "wasm32")))] -mod handover_tests { - use super::*; - use std::sync::atomic::AtomicUsize; - - #[test] - fn handover_pair_pending_keeps_promoted_identity_and_rejects_retired_allow() { - let directory = tempfile::tempdir().unwrap(); - let folder = directory.path().join("notebook"); - std::fs::create_dir(&folder).unwrap(); - let sharing = Sharing { - approve: true, - code: super::super::super::code::format(701, "ABCDEF").unwrap(), - ..Sharing::new("").unwrap() - }; - let room = Room::share(&sharing.code, &sharing.password); - let host = Host::start( - crate::session::Notebook::open(&folder, directory.path().join("cache")) - .unwrap() - .into_storage(), - Hello::new("Notebook owner".into(), None).unwrap(), - sharing, - "Fixture", - None, - None, - || {}, - |_| Ok(()), - ) - .unwrap(); - let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared); - let valid = super::super::handover_tests::promoted_routes( - room, - |source, old, old_line| { - (callback.events)(source, Event::Met(old, old_line)); - }, - |source, old, old_line, new, new_line| { - host.allow(&old.peer).unwrap(); - let no_grant = host.sharing().members.is_empty(); - (callback.events)(source, Event::Met(new, new_line)); - (callback.events)(source, Event::Met(old, old_line)); - (callback.events)(source, Event::Left(old)); - let kept = host - .requests() - .first() - .is_some_and(|request| Arc::ptr_eq(request, new)); - host.stop(); - kept && no_grant - }, - ); - assert!( - valid, - "pairing admitted a retired route or retired Left removed the promoted request" - ); - } - - #[test] - fn handover_pair_retired_met_cannot_persist_another_credential() { - let directory = tempfile::tempdir().unwrap(); - let folder = directory.path().join("notebook"); - std::fs::create_dir(&folder).unwrap(); - let sharing = Sharing { - approve: false, - code: super::super::super::code::format(702, "ABCDEF").unwrap(), - ..Sharing::new("").unwrap() - }; - let room = Room::share(&sharing.code, &sharing.password); - let kept = Arc::new(AtomicUsize::new(0)); - let count = Arc::clone(&kept); - let host = Host::start( - crate::session::Notebook::open(&folder, directory.path().join("cache")) - .unwrap() - .into_storage(), - Hello::new("Notebook owner".into(), None).unwrap(), - sharing, - "Fixture", - None, - None, - || {}, - move |_| { - count.fetch_add(1, Ordering::SeqCst); - Ok(()) - }, - ) - .unwrap(); - let callback = Arc::clone(&host.pairing.lock().unwrap().as_ref().unwrap().shared); - let valid = super::super::handover_tests::promoted_routes( - room, - |_, _, _| {}, - |source, old, old_line, new, new_line| { - (callback.events)(source, Event::Met(new, new_line)); - let after_current = kept.load(Ordering::SeqCst); - (callback.events)(source, Event::Met(old, old_line)); - let refused = kept.load(Ordering::SeqCst) == after_current - && after_current == 2 - && host.sharing().members.len() == 1; - host.stop(); - refused - }, - ); - assert!(valid, "retired pairing Met persisted another credential"); - } - - #[test] - fn handover_retired_membership_rejects_deferred_access_met() { - let directory = tempfile::tempdir().unwrap(); - let folder = directory.path().join("notebook"); - std::fs::create_dir(&folder).unwrap(); - let secret = [71; 16]; - let sharing = Sharing { - code: super::super::super::code::format(703, "ABCDEF").unwrap(), - members: vec![Device { - secret, - name: "Retired device".into(), - device: None, - }], - ..Sharing::new("").unwrap() - }; - let host = Host::start( - crate::session::Notebook::open(&folder, directory.path().join("cache")) - .unwrap() - .into_storage(), - Hello::new("Notebook owner".into(), None).unwrap(), - sharing, - "Fixture", - None, - None, - || {}, - |_| Ok(()), - ) - .unwrap(); - let callback = Arc::clone(&host.members.access.lock().unwrap()[&secret].shared); - let refused = super::super::handover_tests::promoted_routes( - Room::Notebook(secret), - |_, _, _| {}, - |source, _, _, current, line| { - host.remove(&secret).unwrap(); - (callback.events)(source, Event::Met(current, line)); - let refused = host.members.served.guests.lock().unwrap().is_empty() - && host.sharing().members.is_empty(); - host.stop(); - refused - }, - ); - assert!( - refused, - "deferred access callback readmitted a removed device" - ); - } -} +#[path = "handover_tests.rs"] +mod handover_tests; diff --git a/crates/notebook/src/live/web.rs b/crates/notebook/src/live/web.rs index 7d8eca0ceb037b16ef067bdb15349bc77e2aac7b..98fbecb9d6c7715493d012d852c2042a83c44a31 100644 --- a/crates/notebook/src/live/web.rs +++ b/crates/notebook/src/live/web.rs @@ -36,7 +36,9 @@ extern "C" { fn close(id: u32); } -pub struct Live(Arc); +pub struct Live { + shared: Arc, +} struct Shared { me: Hello, @@ -209,35 +211,35 @@ impl Live { } } })?; - Ok(Self(shared)) + Ok(Self { shared }) } pub fn code(&self) -> Option { - match &self.0.room { + match &self.shared.room { Room::Code { code, .. } => Some(code.clone()), _ => None, } } pub fn relayed(&self) -> Relayed { - self.0.state.lock().unwrap().relayed.clone() + self.shared.state.lock().unwrap().relayed.clone() } pub fn failed(&self) -> u32 { - self.0.state.lock().unwrap().failed + self.shared.state.lock().unwrap().failed } pub fn other_version(&self) -> Option { - self.0.state.lock().unwrap().outdated + self.shared.state.lock().unwrap().outdated } pub fn burned(&self) -> bool { - self.0.state.lock().unwrap().burned + self.shared.state.lock().unwrap().burned } pub fn sender(&self) -> Sender { - Sender(Arc::downgrade(&self.0)) + Sender(Arc::downgrade(&self.shared)) } pub fn peers(&self) -> Vec { - self.0.peers() + self.shared.peers() } pub fn line(&self, peer: &[u8; 16]) -> Option { - self.0 + self.shared .state .lock() .unwrap() @@ -247,7 +249,7 @@ impl Live { .map(|s| s.line.clone()) } pub fn set_presence(&self, presence: Presence) { - let mut state = self.0.state.lock().unwrap(); + let mut state = self.shared.state.lock().unwrap(); if state.presence == presence { return; } @@ -256,7 +258,7 @@ impl Live { return; } drop(state); - let weak = Arc::downgrade(&self.0); + let weak = Arc::downgrade(&self.shared); let _ = crate::task::spawn("live presence", move || async move { let (_, wait) = crate::task::channel(); crate::task::wait(&wait, Some(Duration::from_millis(100))).await; @@ -272,7 +274,7 @@ impl Live { } pub fn leave(self, reason: &str) { let lines: Vec<_> = self - .0 + .shared .state .lock() .unwrap() @@ -288,8 +290,8 @@ impl Live { impl Drop for Live { fn drop(&mut self) { - self.0.stopped.store(true, Ordering::Release); - let socket = self.0.state.lock().unwrap().socket; + self.shared.stopped.store(true, Ordering::Release); + let socket = self.shared.state.lock().unwrap().socket; close(socket); } } @@ -310,24 +312,20 @@ impl Shared { } fn gone(&self, hello: &Arc) -> bool { - !self.state.lock().unwrap().streams.values().any(|stream| { - stream - .peer - .as_ref() - .is_some_and(|peer| Arc::ptr_eq(&peer.hello, hello)) - }) - } - - fn connected(&self, hello: &Arc) -> bool { - !self.stopped.load(Ordering::Acquire) - && self.state.lock().unwrap().streams.values().any(|stream| { - stream - .peer - .as_ref() + let state = self.state.lock().unwrap(); + !(state.members.values().map(|member| &member.peer)) + .chain(state.streams.values().map(|stream| &stream.peer)) + .any(|peer| { + peer.as_ref() .is_some_and(|peer| Arc::ptr_eq(&peer.hello, hello)) }) } + /// Whether `hello` is how a peer is in the room now: its stream, or its place in the group. + fn connected(&self, hello: &Arc) -> bool { + !self.stopped.load(Ordering::Acquire) && !self.gone(hello) + } + fn connect(self: &Arc) -> io::Result<()> { let weak = Arc::downgrade(self); let message = Closure::::new(move |data: JsValue| {