diff --git a/crates/notebook/src/live.rs b/crates/notebook/src/live.rs index 69906219840dc6e2a84930fbba5ea4f98f7f6513..acd6687f01d6159b33e77cd431638e78bb0c0495 100644 --- a/crates/notebook/src/live.rs +++ b/crates/notebook/src/live.rs @@ -61,7 +61,8 @@ struct Shared { secret: Vec, reach: Option, state: Mutex, - events: Box, + #[allow(clippy::type_complexity)] + events: Box, Event) + Send + Sync>, stopped: AtomicBool, connections: AtomicU64, } @@ -231,6 +232,16 @@ impl Live { reach: Option, relay: Option<&str>, events: impl Fn(Event) + Send + Sync + 'static, + ) -> io::Result { + Self::start_with(me, room, reach, relay, move |_, event| events(event)) + } + + fn start_with( + me: Hello, + room: &Room, + reach: Option, + relay: Option<&str>, + events: impl Fn(&Arc, Event) + Send + Sync + 'static, ) -> io::Result { let host = match reach { Some(Reach::Network) => IpAddr::V4(Ipv4Addr::UNSPECIFIED), @@ -440,6 +451,21 @@ impl Drop for Live { } impl Shared { + fn gone(&self, hello: &Arc) -> bool { + !self.state.lock().unwrap().peers.contains_key(&hello.peer) + } + + fn connected(&self, hello: &Arc) -> bool { + !self.stopped.load(Ordering::Acquire) + && self + .state + .lock() + .unwrap() + .peers + .get(&hello.peer) + .is_some_and(|link| Arc::ptr_eq(&link.peer.hello, hello)) + } + /// The peers in the room: those met directly, and the rest as the relay's group or a /// stream through the relay last heard of them. fn peers(&self) -> Vec { @@ -547,7 +573,7 @@ impl Shared { /// Counts a meeting that failed on the secret: the end sharing a code burns it after too /// many, and an end that typed one gives up at once. - fn failed(&self) { + fn failed(self: &Arc) { let mut state = self.state.lock().unwrap(); state.failed += 1; match &self.room { @@ -556,7 +582,7 @@ impl Shared { _ => {} } drop(state); - (self.events)(Event::Changed); + (self.events)(self, Event::Changed); } /// Meets the peer at the other end of `pipe` in room `tag`, then reads from it until it @@ -597,7 +623,7 @@ impl Shared { if matches!(self.room, Room::Code { owner: false, .. }) { self.stopped.store(true, Ordering::Release); } - (self.events)(Event::Changed); + (self.events)(&self, Event::Changed); } else if error.kind() == io::ErrorKind::InvalidData { pipe.met(false); self.failed(); @@ -614,17 +640,14 @@ impl Shared { let connection = self.connections.fetch_add(1, Ordering::Relaxed); let (out, outgoing) = mpsc::channel(); let line = Line(out); - { + let retired = { let mut state = self.state.lock().unwrap(); // A peer met both directly and through a relay keeps the direct connection, as // both ends then agree. let kept = match state.peers.get(&peer) { _ if peer == self.me.peer => false, Some(link) if link.pipe.direct() || !pipe.direct() => false, - Some(link) => { - link.pipe.shutdown(); - true - } + Some(_) => true, None => true, }; if !kept { @@ -632,9 +655,6 @@ impl Shared { pipe.shutdown(); return false; } - if self.carries_presence(&*pipe) { - let _ = line.0.send(Out::Presence); - } state.peers.insert( peer, Link { @@ -646,13 +666,19 @@ impl Shared { line: line.clone(), pipe: Arc::clone(&pipe), }, - ); + ) + }; + if let Some(retired) = retired { + retired.pipe.shutdown(); + } + if self.carries_presence(&*pipe) { + let _ = line.0.send(Out::Presence); } let writing = Arc::clone(&pipe); let shared = Arc::clone(&self); thread::spawn(move || shared.write(&*writing, send, outgoing)); - (self.events)(Event::Met(&hello, &line)); - (self.events)(Event::Changed); + (self.events)(&self, Event::Met(&hello, &line)); + (self.events)(&self, Event::Changed); let ended = loop { let (message, body) = match receive.receive(&mut stream) { Ok(frame) => frame, @@ -664,17 +690,22 @@ impl Shared { let Ok(presence) = minicbor::decode::(&body) else { break None; }; - if let Some(link) = self.state.lock().unwrap().peers.get_mut(&peer) { + if let Some(link) = self.state.lock().unwrap().peers.get_mut(&peer) + && link.connection == connection + { link.peer.presence = Some(presence); } - (self.events)(Event::Changed); + (self.events)(&self, Event::Changed); } kind => { - (self.events)(Event::Frame { - from: &hello, - kind, - body: &body, - }); + (self.events)( + &self, + Event::Frame { + from: &hello, + kind, + body: &body, + }, + ); // The peer hangs up after its bye. if kind == kind::BYE { break None; @@ -695,8 +726,8 @@ impl Shared { { state.peers.remove(&peer); drop(state); - (self.events)(Event::Left(&hello)); - (self.events)(Event::Changed); + (self.events)(&self, Event::Left(&hello)); + (self.events)(&self, Event::Changed); } true } diff --git a/crates/notebook/src/live/relay.rs b/crates/notebook/src/live/relay.rs index cbc944cbd9da0bb0dad79b549f3e83c57d4687d8..92badc5607b271b115ac7c5522da9e5a6f0c942f 100644 --- a/crates/notebook/src/live/relay.rs +++ b/crates/notebook/src/live/relay.rs @@ -44,12 +44,12 @@ pub(super) fn join(shared: &Arc, url: &str, port: u16) -> io::Result<()> } /// Records how the relay answered, telling `shared`'s events where it changed. -fn answered(shared: &Shared, answer: Answer) { +fn answered(shared: &Arc, answer: Answer) { let mut state = shared.state.lock().unwrap(); if state.relayed.as_ref() != Some(&answer) { state.relayed = Some(answer); drop(state); - (shared.events)(Event::Changed); + (shared.events)(shared, Event::Changed); } } @@ -245,7 +245,7 @@ fn session( state.code = Some(code); drop(state); shared.advertise(port); - (shared.events)(Event::Changed); + (shared.events)(shared, Event::Changed); } } Ok(Notice::Welcome { you, members }) => { @@ -263,14 +263,14 @@ fn session( let left = (socket.group.as_ref()) .and_then(|group| group.members.lock().unwrap().remove(&slot)); if left.is_some() { - (shared.events)(Event::Changed); + (shared.events)(shared, Event::Changed); } } Ok(Notice::Burned) => { // Coming back, ask for another number. *nameplate = None; shared.state.lock().unwrap().burned = true; - (shared.events)(Event::Changed); + (shared.events)(shared, Event::Changed); } Err(()) => {} }, @@ -296,7 +296,7 @@ fn session( if let Some(group) = &socket.group { *group.out.lock().unwrap() = None; group.members.lock().unwrap().clear(); - (shared.events)(Event::Changed); + (shared.events)(shared, Event::Changed); } } @@ -412,7 +412,7 @@ fn heard(shared: &Arc, socket: &Arc, tag: Option<&str>, slot: u3 if serves { meet(shared, socket, tag, slot, None); } - (shared.events)(Event::Changed); + (shared.events)(shared, Event::Changed); } kind::PRESENCE => { let Ok(presence) = minicbor::decode::(&body) else { @@ -422,17 +422,20 @@ fn heard(shared: &Arc, socket: &Arc, tag: Option<&str>, slot: u3 if let Some(peer) = members.get_mut(&slot).and_then(|m| m.peer.as_mut()) { peer.presence = Some(presence); drop(members); - (shared.events)(Event::Changed); + (shared.events)(shared, Event::Changed); } } // A peer met directly says the rest there. kind if kind < 256 => { if let Some(hello) = hello.filter(|hello| !shared.direct(&hello.peer)) { - (shared.events)(Event::Frame { - from: &hello, - kind, - body: &body, - }); + (shared.events)( + shared, + Event::Frame { + from: &hello, + kind, + body: &body, + }, + ); } } _ => {} diff --git a/crates/notebook/src/live/share.rs b/crates/notebook/src/live/share.rs index 5f2dc1103bd0fd976b6b0521565b0b8ca451427d..cfbd6dbf93113a9dcf1ea09124ea02b48bc75815 100644 --- a/crates/notebook/src/live/share.rs +++ b/crates/notebook/src/live/share.rs @@ -18,7 +18,7 @@ use std::{ io, path::PathBuf, sync::{ - Arc, Condvar, Mutex, + Arc, Condvar, Mutex, Weak, atomic::{AtomicBool, AtomicU64, Ordering}, mpsc, }, @@ -29,6 +29,9 @@ 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")] @@ -304,6 +307,7 @@ struct Served { /// many more it may start now, and the sections it read lately, newest first, as it keeps /// their images. struct Admitted { + hello: Arc, line: Line, queue: mpsc::SyncSender<(u16, Vec)>, starts: f64, @@ -336,9 +340,32 @@ fn refused(kind: io::ErrorKind, message: &str) -> Error { impl Served { /// Serves `peer` on `line`, through workers of its own that end as it leaves. - fn admit(self: &Arc, peer: [u8; 16], line: &Line) { + fn admit( + self: &Arc, + source: &Arc, + hello: &Arc, + line: &Line, + ) -> bool { + let peer = hello.peer; let (queue, waiting) = mpsc::sync_channel::<(u16, Vec)>(QUEUED - WORKERS); let waiting = Arc::new(Mutex::new(waiting)); + let guest = Admitted { + hello: Arc::clone(hello), + line: line.clone(), + queue, + starts: BURST, + counted: Instant::now(), + held: Vec::new(), + }; + let mut guests = self.guests.lock().unwrap(); + if !source.connected(hello) { + drop(guests); + drop(guest); + return false; + } + let retired = guests.insert(peer, guest); + drop(guests); + drop(retired); for _ in 0..WORKERS { let (served, waiting, line) = (Arc::downgrade(self), Arc::clone(&waiting), line.clone()); @@ -352,23 +379,20 @@ impl Served { } }); } - let guest = Admitted { - line: line.clone(), - queue, - starts: BURST, - counted: Instant::now(), - held: Vec::new(), - }; - self.guests.lock().unwrap().insert(peer, guest); + true } /// Hands a request from `peer` to its workers, or hangs up on a guest that has too many /// waiting or starts them too fast. - fn queue(&self, peer: &[u8; 16], kind: u16, body: &[u8]) { + fn queue(&self, source: &Arc, hello: &Arc, kind: u16, body: &[u8]) { + let peer = &hello.peer; let mut guests = self.guests.lock().unwrap(); let Some(guest) = guests.get_mut(peer) else { return; }; + if !Arc::ptr_eq(&guest.hello, hello) || !source.connected(hello) { + return; + } let continued = kind == kind::PUT || matches!(kind, kind::READ | kind::READ_FILE) && minicbor::decode::(body).is_ok_and(|request| request.handle.is_some()); @@ -380,21 +404,33 @@ impl Served { guest.starts -= 1.0; } if guest.starts < 0.0 || guest.queue.try_send((kind, body.to_vec())).is_err() { - guest.line.hang_up(FLOODED); - guests.remove(peer); + let guest = guests.remove(peer); + drop(guests); + if let Some(guest) = guest { + guest.line.hang_up(FLOODED); + } } } - fn forget(&self, peer: &[u8; 16]) { - self.guests.lock().unwrap().remove(peer); - self.snapshots - .lock() - .unwrap() - .retain(|(guest, _), _| guest != peer); - self.puts - .lock() - .unwrap() - .retain(|(guest, _), _| guest != peer); + fn forget(&self, peer: &[u8; 16], expected: Option<(&Arc, &Arc)>) { + let retired = { + let mut snapshots = self.snapshots.lock().unwrap(); + let mut puts = self.puts.lock().unwrap(); + let mut guests = self.guests.lock().unwrap(); + if let Some((source, hello)) = expected + && (!source.gone(hello) + || !guests + .get(peer) + .is_some_and(|guest| Arc::ptr_eq(&guest.hello, hello))) + { + return; + } + let retired = guests.remove(peer); + snapshots.retain(|(guest, _), _| guest != peer); + puts.retain(|(guest, _), _| guest != peer); + retired + }; + drop(retired); } /// Tells every guest, and the host, that a guest changed the files at `paths`. @@ -1086,10 +1122,16 @@ impl Guest { catalog: Mutex::default(), }); let heard = Arc::clone(&inner); - let live = Live::start(me, &Room::Notebook(secret), reach, relay, move |event| { - heard.heard(event); - (heard.events)(); - })?; + let live = Live::start_with( + me, + &Room::Notebook(secret), + reach, + relay, + move |source, event| { + heard.heard(source, event); + (heard.events)(); + }, + )?; Ok(Arc::new(Self { live, inner })) } @@ -1565,148 +1607,268 @@ impl Inner { } } - fn heard(self: &Arc, event: Event) { - let serves = |hello: &Hello| { - self.host - .lock() - .unwrap() - .as_ref() - .is_some_and(|(host, _)| host.peer == hello.peer) - }; + fn heard(self: &Arc, source: &Arc, event: Event) { match event { Event::Met(hello, line) if hello.serves == Some(self.share) => { - *self.host.lock().unwrap() = Some((Arc::clone(hello), line.clone())); + let mut host = self.host.lock().unwrap(); + let retired = if source.connected(hello) { + host.replace((Arc::clone(hello), line.clone())) + } else { + None + }; + drop(host); + drop(retired); } - Event::Left(hello) if serves(hello) => { - *self.host.lock().unwrap() = None; - drop(self.presence.lock().unwrap().take()); - self.current.lock().unwrap().clear(); - // Each request waiting hears its answer was lost. - self.pending.lock().unwrap().clear(); - if let Some(reports) = self.watch.lock().unwrap().take() { + Event::Left(hello) => { + let (retired, presence, pending, reports) = { + let mut host = self.host.lock().unwrap(); + if !host + .as_ref() + .is_some_and(|(admitted, _)| Arc::ptr_eq(admitted, hello)) + || !source.gone(hello) + { + return; + } + let retired = host.take(); + let presence = self.presence.lock().unwrap().take(); + self.current.lock().unwrap().clear(); + let pending = std::mem::take(&mut *self.pending.lock().unwrap()); + let reports = self.watch.lock().unwrap().take(); + (retired, presence, pending, reports) + }; + drop(retired); + drop(presence); + drop(pending); + if let Some(reports) = reports { reports.lost(); } } - Event::Frame { - from, kind, body, .. - } => match kind { - kind::WELCOME if serves(from) => { - if let Ok(welcome) = minicbor::decode::(body) { - self.join_presence(welcome.room); - } + Event::Frame { from, kind, body } => { + let host = self.host.lock().unwrap(); + if !host + .as_ref() + .is_some_and(|(admitted, _)| Arc::ptr_eq(admitted, from)) + || !source.connected(from) + { + return; } - kind::REPLY if serves(from) => { - if let Ok(reply) = minicbor::decode::(body) - && let Some(waiting) = self.pending.lock().unwrap().remove(&reply.id) - { - let _ = waiting.send(reply); + match kind { + kind::WELCOME => { + if let Ok(welcome) = minicbor::decode::(body) { + drop(host); + self.join_presence(source, from, welcome.room); + } } - } - kind::DELTA if serves(from) => { - if let Ok(delta) = minicbor::decode::(body) { - self.apply(&delta); + 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::TOUCHED if serves(from) => { - if let Ok(touched) = minicbor::decode::(body) { - self.touched(&touched); - let reports = self.watch.lock().unwrap().clone(); - if let Some(reports) = reports { - reports.touched(&touched.paths); + kind::DELTA => { + if let Ok(delta) = minicbor::decode::(body) { + self.apply(&delta); } } - } - kind::BYE if serves(from) => { - if let Ok(bye) = minicbor::decode::(body) { - let ended = match bye.reason.as_str() { - STOPPED => Some(Ended::Stopped), - "removed" => Some(Ended::Removed), - _ => None, - }; - if ended.is_some() { - *self.ended.lock().unwrap() = ended; + kind::TOUCHED => { + if let Ok(touched) = minicbor::decode::(body) { + self.touched(&touched); + let reports = self.watch.lock().unwrap().clone(); + drop(host); + if let Some(reports) = reports { + reports.touched(&touched.paths); + } + } + } + kind::BYE => { + if let Ok(bye) = minicbor::decode::(body) { + let ended = match bye.reason.as_str() { + STOPPED => Some(Ended::Stopped), + "removed" => Some(Ended::Removed), + _ => None, + }; + if ended.is_some() { + *self.ended.lock().unwrap() = ended; + } } } + _ => {} } - _ => {} - }, + } _ => {} } } - fn join_presence(self: &Arc, secret: [u8; 16]) { - let mut presence = self.presence.lock().unwrap(); - if presence.as_ref().is_some_and(|(held, _)| *held == secret) { - return; + fn join_presence( + self: &Arc, + source: &Arc, + 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( + let live = Live::start_with( self.me.clone(), &Room::Notebook(secret), self.reach, self.relay.as_deref(), - move |event| { - if let Some(inner) = inner.upgrade() { - if matches!(event, Event::Changed) { - let mut connected = connected.lock().unwrap(); - let host = inner - .host - .lock() - .unwrap() - .as_ref() - .map(|(host, _)| host.peer); - let peers = inner - .presence - .lock() - .unwrap() - .as_ref() - .map(|(_, live)| live.peers()) - .unwrap_or_default(); - let present = - host.filter(|host| peers.iter().any(|peer| peer.hello.peer == *host)); - if *connected != present { - *connected = present; - inner.current.lock().unwrap().clear(); - drop(connected); - let reports = inner.watch.lock().unwrap().clone(); - if present.is_some() - && let Some(reports) = reports - { - reports.touched(&[String::new()]); + move |source, event| { + let Some(inner) = inner.upgrade() else { + return; + }; + let reports = { + let host = inner.host.lock().unwrap(); + let presence = inner.presence.lock().unwrap(); + 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 { + return; + } + match event { + Event::Changed => { + let mut connected = connected.lock().unwrap(); + let host = host.as_ref().map(|(host, _)| host.peer); + let peers = source.peers(); + let present = host + .filter(|host| peers.iter().any(|peer| peer.hello.peer == *host)); + if *connected == present { + None + } else { + *connected = present; + inner.current.lock().unwrap().clear(); + if present.is_some() { + inner + .watch + .lock() + .unwrap() + .clone() + .map(|reports| (reports, vec![String::new()])) + } else { + None + } } } - } - if matches!( - event, - Event::Frame { - kind: kind::DELTA | kind::TOUCHED, - .. + Event::Frame { from, kind, body } + if host + .as_ref() + .is_some_and(|(host, _)| host.peer == from.peer) + && source + .peers() + .iter() + .any(|peer| Arc::ptr_eq(&peer.hello, from)) => + { + match kind { + kind::DELTA => { + if let Ok(delta) = minicbor::decode::(body) { + inner.apply(&delta); + } + None + } + kind::TOUCHED => { + if let Ok(touched) = minicbor::decode::(body) { + inner.touched(&touched); + inner + .watch + .lock() + .unwrap() + .clone() + .map(|reports| (reports, touched.paths)) + } else { + None + } + } + _ => None, + } } - ) { - inner.heard(event); + _ => None, } - (inner.events)(); - } - }, - ); - match live { - Ok(live) => { - live.set_presence(self.here.lock().unwrap().clone()); - *presence = Some((secret, live)); - drop(presence); - let mut current = self.current.lock().unwrap(); - let paths: Vec = current.keys().cloned().collect(); - current.clear(); - drop(current); - let reports = self.watch.lock().unwrap().clone(); - if let Some(reports) = reports { + }; + if let Some((reports, paths)) = reports { reports.touched(&paths); } - } + (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 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(); + if host + .as_ref() + .is_some_and(|(admitted, _)| Arc::ptr_eq(admitted, from)) + && source.connected(from) + { + let mut presence = self.presence.lock().unwrap(); + if presence.as_ref().is_some_and(|(held, _)| *held == secret) { + (None, Vec::new(), None) + } else { + let retired = presence.replace((secret, candidate.take().unwrap())); + let paths = self + .current + .lock() + .unwrap() + .drain() + .map(|(path, _)| path) + .collect::>(); + let reports = self.watch.lock().unwrap().clone(); + (retired, paths, reports) + } + } else { + (None, Vec::new(), None) + } + }; + drop(here); + drop(retired); + if candidate.is_none() { + (refresh.events)(&refresh, Event::Changed); + } + drop(candidate); + if let Some(reports) = reports { + reports.touched(&paths); + } + } } /// A section a Live Share host serves, as a guest's replica publishes to it. diff --git a/crates/notebook/src/live/share/batch.rs b/crates/notebook/src/live/share/batch.rs index 9830ffaa2c6d86e2145bcfe8bee15222ecb4e370..d7d95501d9a7549fc0db3a68411beb4d32a2f005 100644 --- a/crates/notebook/src/live/share/batch.rs +++ b/crates/notebook/src/live/share/batch.rs @@ -342,6 +342,10 @@ mod tests { served.guests.lock().unwrap().insert( peer, Admitted { + hello: Arc::new(Hello { + peer, + ..Hello::new("Guest".into(), None).unwrap() + }), line: crate::live::Line(line), queue, starts: BURST, diff --git a/crates/notebook/src/live/share/handover_tests.rs b/crates/notebook/src/live/share/handover_tests.rs new file mode 100644 index 0000000000000000000000000000000000000000..564dc26ff817a4108a82b23ce37640ae2a02c8e2 --- /dev/null +++ b/crates/notebook/src/live/share/handover_tests.rs @@ -0,0 +1,721 @@ +use super::super::{GONE, OPENING, Pipe, Shared, State}; +use super::*; +use std::{ + io::{Read, Write}, + net::{Shutdown, TcpListener, TcpStream}, +}; + +struct Barrier { + reached: mpsc::Sender<()>, + resume: mpsc::Receiver<()>, +} + +impl Barrier { + fn wait(self) -> io::Result<()> { + self.reached.send(()).map_err(io::Error::other)?; + self.resume.recv_timeout(GONE).map_err(io::Error::other) + } +} + +fn barrier() -> (Barrier, mpsc::Receiver<()>, mpsc::Sender<()>) { + let (reached, waiting) = mpsc::channel(); + let (resume, continued) = mpsc::channel(); + ( + Barrier { + reached, + resume: continued, + }, + waiting, + resume, + ) +} + +struct HandoverPipe { + stream: TcpStream, + direct: bool, + body: Mutex>, + retired: mpsc::Sender<()>, +} + +impl Pipe for HandoverPipe { + fn read(&self, buffer: &mut [u8]) -> io::Result { + let gate = if buffer.len() > 4 { + self.body.lock().unwrap().take() + } else { + None + }; + if let Some(gate) = gate { + // Preserve a complete copied ciphertext block across the old pipe's shutdown. + Read::read_exact(&mut &self.stream, buffer)?; + gate.wait()?; + Ok(buffer.len()) + } else { + Read::read(&mut &self.stream, buffer) + } + } + + fn write(&self, bytes: &[u8]) -> io::Result { + Write::write(&mut &self.stream, bytes) + } + + fn set_read_timeout(&self, timeout: Duration) -> io::Result<()> { + self.stream.set_read_timeout(Some(timeout)) + } + + fn shutdown(&self) { + let _ = self.stream.shutdown(Shutdown::Both); + } + + fn direct(&self) -> bool { + self.direct + } +} + +impl Drop for HandoverPipe { + fn drop(&mut self) { + let _ = self.retired.send(()); + } +} + +struct Handover { + shared: Arc, + guest: Arc, + pipes: Vec>, + readers: Vec<(mpsc::Receiver<()>, thread::JoinHandle)>, + retired: (mpsc::Sender<()>, mpsc::Receiver<()>), + observed: mpsc::Receiver<()>, + releases: Vec>, +} + +impl Handover { + fn new(met: Option) -> Self { + let me = Hello::new("Guest".into(), None).unwrap(); + let guest = Arc::new(Inner { + share: [41; 16], + me: me.clone(), + reach: None, + relay: None, + events: Arc::new(|| {}), + presence: Mutex::default(), + here: Mutex::default(), + host: Mutex::default(), + pending: Mutex::default(), + next: AtomicU64::new(1), + watch: Mutex::default(), + ended: Mutex::default(), + asked: Default::default(), + images: Mutex::default(), + current: Mutex::default(), + }); + let heard = Arc::clone(&guest); + let met = Mutex::new(met); + let (events, observed) = mpsc::channel(); + let room = Room::Notebook([42; 16]); + let shared = Arc::new(Shared { + me, + secret: room.secret(), + room, + reach: None, + state: Mutex::new(State::default()), + events: Box::new(move |source, event| { + let meeting = matches!(&event, Event::Met(..)); + if meeting { + let gate = met.lock().unwrap().take(); + if let Some(gate) = gate { + gate.wait().unwrap(); + } + } + heard.heard(source, event); + if meeting { + let _ = events.send(()); + } + }), + stopped: AtomicBool::new(false), + connections: AtomicU64::new(0), + }); + Self { + shared, + guest, + pipes: Vec::new(), + readers: Vec::new(), + retired: mpsc::channel(), + observed, + releases: Vec::new(), + } + } + + fn open(&mut self, host: &Hello, direct: bool) -> (TcpStream, wire::Sealer) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let mut remote = + TcpStream::connect_timeout(&listener.local_addr().unwrap(), OPENING).unwrap(); + remote.set_read_timeout(Some(OPENING)).unwrap(); + remote.set_write_timeout(Some(OPENING)).unwrap(); + let (local, _) = listener.accept().unwrap(); + local.set_write_timeout(Some(OPENING)).unwrap(); + let pipe = Arc::new(HandoverPipe { + stream: local, + direct, + body: Mutex::default(), + retired: self.retired.0.clone(), + }); + self.pipes.push(Arc::clone(&pipe)); + let shared = Arc::clone(&self.shared); + let tag = shared.room.tag().unwrap(); + let reader_tag = tag.clone(); + let (finished, done) = mpsc::channel(); + let reader = thread::spawn(move || { + let retry = shared.run(pipe, wire::Side::Responder, &reader_tag); + let _ = finished.send(()); + retry + }); + self.readers.push((done, reader)); + let (mut send, mut receive) = wire::open( + &mut remote, + wire::Side::Initiator, + &tag, + &self.shared.secret, + ) + .unwrap(); + send.send(&mut remote, kind::HELLO, host).unwrap(); + let (message, body) = receive.receive(&mut remote).unwrap(); + assert_eq!(message, kind::HELLO); + assert_eq!(minicbor::decode::(&body).unwrap(), self.shared.me); + (remote, send) + } + + fn heard(&self) { + self.observed.recv_timeout(OPENING).unwrap(); + } +} + +impl Drop for Handover { + fn drop(&mut self) { + self.shared.stopped.store(true, Ordering::Release); + for release in self.releases.drain(..) { + let _ = release.send(()); + } + for pipe in &self.pipes { + 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()); + 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(); + } + } + 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" + ); + } + } +} + +#[test] +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); + fixture.heard(); + let (gate, copied, resume) = barrier(); + fixture.releases.push(resume.clone()); + *fixture.pipes[0].body.lock().unwrap() = Some(gate); + send.send( + &mut old, + kind::BYE, + &wire::Bye { + reason: STOPPED.into(), + }, + ) + .unwrap(); + copied.recv_timeout(OPENING).unwrap(); + host.name = "Promoted host route".into(); + let (_new, _) = fixture.open(&host, true); + fixture.heard(); + assert!( + fixture.shared.state.lock().unwrap().peers[&host.peer] + .pipe + .direct() + ); + assert!(fixture.guest.ended.lock().unwrap().is_none()); + resume.send(()).unwrap(); + 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_stale_met_cannot_replace_promoted_guest_line() { + let (gate, delayed, resume) = barrier(); + let mut fixture = Handover::new(Some(gate)); + 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); + 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}"), + }; + assert_eq!( + fixture.shared.state.lock().unwrap().peers[&host.peer] + .peer + .hello + .name, + host.name + ); + let selected = fixture + .guest + .host + .lock() + .unwrap() + .as_ref() + .unwrap() + .0 + .name + .clone(); + drop(fixture); + println!("HANDOVER_MET_PROMOTION_OVERLAP={overlap}"); + assert_eq!( + selected, host.name, + "stale Met replaced the promoted guest Line" + ); +} + +impl Handover { + fn intercept( + &mut self, + before: impl Fn(&Arc, &Event<'_>) + Send + Sync + 'static, + ) -> mpsc::Receiver { + let guest = Arc::clone(&self.guest); + let (observed, waiting) = mpsc::channel(); + Arc::get_mut(&mut self.shared).unwrap().events = Box::new(move |source, event| { + let kind = match &event { + Event::Met(..) => kind::HELLO, + Event::Left(..) => kind::BYE, + Event::Frame { kind, .. } => *kind, + Event::Changed => 0, + }; + before(source, &event); + guest.heard(source, event); + let _ = observed.send(kind); + }); + waiting + } +} + +fn observed(events: &mpsc::Receiver, wanted: u16) { + let end = std::time::Instant::now() + OPENING; + loop { + let kind = events + .recv_timeout(end.saturating_duration_since(std::time::Instant::now())) + .unwrap(); + if kind == wanted { + return; + } + } +} + +#[test] +fn handover_old_left_preserves_promoted_host_and_pending_reply() { + let mut fixture = Handover::new(None); + let (gate, removed, resume) = barrier(); + fixture.releases.push(resume.clone()); + let gate = Mutex::new(Some(gate)); + let events = fixture.intercept(move |_, event| { + if matches!(event, Event::Left(_)) + && let Some(gate) = gate.lock().unwrap().take() + { + gate.wait().unwrap(); + } + }); + let mut host = Hello::new("Old host route".into(), None).unwrap(); + host.serves = Some(fixture.guest.share); + 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); + observed(&events, kind::HELLO); + let (answer, waiting) = crate::task::response::(); + fixture.guest.pending.lock().unwrap().insert(73, answer); + resume.send(()).unwrap(); + fixture.readers[0].0.recv_timeout(OPENING).unwrap(); + let host_kept = fixture + .guest + .host + .lock() + .unwrap() + .as_ref() + .is_some_and(|(hello, _)| hello.name == host.name); + let pending_kept = fixture.guest.pending.lock().unwrap().contains_key(&73); + let answer_kept = matches!(waiting.try_recv(), Err(mpsc::TryRecvError::Empty)); + drop(fixture); + assert!( + host_kept && pending_kept && answer_kept, + "retired Left cleared the promoted host or pending reply" + ); +} + +#[test] +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); + fixture.heard(); + let (answer, waiting) = crate::task::response::(); + fixture.guest.pending.lock().unwrap().insert(74, answer); + remote.shutdown(Shutdown::Both).unwrap(); + fixture.readers[0].0.recv_timeout(OPENING).unwrap(); + let host_cleared = fixture.guest.host.lock().unwrap().is_none(); + let pending_cleared = fixture.guest.pending.lock().unwrap().is_empty(); + let answer_lost = matches!(waiting.try_recv(), Err(mpsc::TryRecvError::Disconnected)); + drop(fixture); + assert!( + host_cleared && pending_cleared && answer_lost, + "selected route loss retained the host or pending reply" + ); +} + +fn served() -> (tempfile::TempDir, Arc) { + let directory = tempfile::tempdir().unwrap(); + let folder = directory.path().join("notebook"); + std::fs::create_dir(&folder).unwrap(); + let served = Arc::new(Served { + storage: crate::session::Notebook::open(&folder, directory.path().join("cache")) + .unwrap() + .into_storage(), + images: Mutex::default(), + snapshots: Mutex::default(), + puts: Mutex::default(), + guests: Mutex::default(), + writers: Mutex::default(), + host: Mutex::default(), + room: Mutex::default(), + }); + (directory, served) +} + +#[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::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); + observed(&events, kind::HELLO); + let (gate, copied, resume) = barrier(); + fixture.releases.push(resume.clone()); + *fixture.pipes[0].body.lock().unwrap() = Some(gate); + send.send( + &mut old, + kind::PUT, + &Request { + id: 80, + bytes: Some(vec![1, 2, 3]), + ..Request::default() + }, + ) + .unwrap(); + copied.recv_timeout(OPENING).unwrap(); + host.name = "Promoted route".into(); + let (mut new, mut send) = 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, + 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; + served.guests.lock().unwrap().clear(); + drop(fixture); + assert!( + no_stale_request && accepted, + "retired request entered the promoted queue or current request was lost" + ); +} + +#[test] +fn handover_old_left_preserves_promoted_host_uploads_and_snapshots() { + let (_directory, served) = served(); + let mut fixture = Handover::new(None); + let (gate, removed, resume) = barrier(); + 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::Left(hello) => { + if let Some(gate) = gate.lock().unwrap().take() { + gate.wait().unwrap(); + } + serving.forget(&hello.peer, Some((source, hello))); + } + _ => {} + }); + let mut host = Hello::new("Old route".into(), None).unwrap(); + host.serves = Some(fixture.guest.share); + 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); + observed(&events, kind::HELLO); + served + .puts + .lock() + .unwrap() + .insert((host.peer, 90), vec![7, 8, 9]); + served + .snapshots + .lock() + .unwrap() + .insert((host.peer, 91), (Arc::new(vec![10, 11]), Instant::now())); + resume.send(()).unwrap(); + fixture.readers[0].0.recv_timeout(OPENING).unwrap(); + let guest_kept = served + .guests + .lock() + .unwrap() + .get(&host.peer) + .is_some_and(|guest| guest.hello.name == host.name); + let upload_kept = served.puts.lock().unwrap().get(&(host.peer, 90)) == Some(&vec![7, 8, 9]); + let snapshot_kept = served + .snapshots + .lock() + .unwrap() + .get(&(host.peer, 91)) + .is_some_and(|(bytes, _)| **bytes == vec![10, 11]); + served.guests.lock().unwrap().clear(); + drop(fixture); + assert!( + guest_kept && upload_kept && snapshot_kept, + "retired Left removed promoted host resources" + ); +} + +#[test] +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); + 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); + let here = here_owner.here.lock().unwrap(); + let (entered, started) = mpsc::channel(); + let (finished, done) = mpsc::channel(); + let installer = thread::spawn(move || { + entered.send(()).unwrap(); + guest.install_presence(&source, &from, secret, live); + 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); + fixture.heard(); + drop(here); + fixture.readers[1].0.recv_timeout(OPENING).unwrap(); + let refused = fixture.guest.presence.lock().unwrap().is_none(); + let promoted = fixture + .guest + .host + .lock() + .unwrap() + .as_ref() + .is_some_and(|(hello, _)| hello.name == host.name); + drop(fixture); + assert!( + refused && promoted, + "deferred old WELCOME installed its presence room" + ); +} + +#[test] +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); + 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(); + let latest = Presence { + section: Some([63; 16]), + ..Presence::default() + }; + let (entered, started) = mpsc::channel(); + let (finished, done) = mpsc::channel(); + let installer = thread::spawn(move || { + entered.send(()).unwrap(); + guest.install_presence(&source, &from, secret, live); + let _ = finished.send(()); + true + }); + fixture.readers.push((done, installer)); + started.recv_timeout(OPENING).unwrap(); + *here = latest.clone(); + drop(here); + fixture.readers[1].0.recv_timeout(OPENING).unwrap(); + let seeded = fixture + .guest + .presence + .lock() + .unwrap() + .as_ref() + .is_some_and(|(held, live)| { + *held == secret && live.shared.state.lock().unwrap().presence == latest + }); + let installed = fixture.guest.presence.lock().unwrap().take(); + drop(installed); + drop(fixture); + assert!( + seeded, + "presence installation overwrote current here with a stale seed" + ); +} + +pub(super) fn promoted_routes( + room: Room, + before: impl FnOnce(&Arc, &Arc, &Line), + check: impl FnOnce(&Arc, &Arc, &Line, &Arc, &Line) -> T, +) -> T { + let mut fixture = Handover::new(None); + let shared = Arc::get_mut(&mut fixture.shared).unwrap(); + shared.secret = room.secret(); + shared.room = room; + let host = Hello::new("Pairing device".into(), None).unwrap(); + let (_old, _) = fixture.open(&host, false); + fixture.heard(); + let (old, old_line) = { + let state = fixture.shared.state.lock().unwrap(); + let link = &state.peers[&host.peer]; + (Arc::clone(&link.peer.hello), link.line.clone()) + }; + before(&fixture.shared, &old, &old_line); + let (_new, _) = fixture.open(&host, true); + fixture.heard(); + let (new, new_line) = { + let state = fixture.shared.state.lock().unwrap(); + let link = &state.peers[&host.peer]; + (Arc::clone(&link.peer.hello), link.line.clone()) + }; + let result = check(&fixture.shared, &old, &old_line, &new, &new_line); + drop((old_line, new_line)); + drop(fixture); + result +} diff --git a/crates/notebook/src/live/share/membership.rs b/crates/notebook/src/live/share/membership.rs index 8e0c550303dd5e4a63c8adfa81751eaf163ce29e..8b21f1fc64358601653b55680f7d45e1fa6d9ae6 100644 --- a/crates/notebook/src/live/share/membership.rs +++ b/crates/notebook/src/live/share/membership.rs @@ -13,7 +13,7 @@ struct Members { relay: Option, sharing: Mutex, changes: Mutex<()>, - pending: Mutex, Line)>>, + pending: Mutex, Line, Weak)>>, access: Mutex>, room: Mutex>, served: Arc, @@ -134,7 +134,7 @@ impl Host { .lock() .unwrap() .values() - .map(|(hello, _)| Arc::clone(hello)) + .map(|(hello, _, _)| Arc::clone(hello)) .collect() } @@ -155,14 +155,27 @@ impl Host { } pub fn allow(&self, peer: &[u8; 16]) -> io::Result<()> { - let pending = self.members.pending.lock().unwrap().remove(peer); - if let Some((hello, line)) = pending { + let request = { + 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(()); + } + pending.remove(peer) + }; + if let Some((hello, line, source)) = request { if let Err(error) = self.members.grant(&hello, &line) { - self.members - .pending - .lock() - .unwrap() - .insert(*peer, (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)(); @@ -171,7 +184,8 @@ impl Host { } pub fn decline(&self, peer: &[u8; 16]) { - if let Some((_, line)) = self.members.pending.lock().unwrap().remove(peer) { + let request = self.members.pending.lock().unwrap().remove(peer); + if let Some((_, line, _)) = request { let _ = line.send(kind::APPROVAL, &wire::Approval::Declined); (self.members.events)(); } @@ -198,20 +212,35 @@ impl Host { let removed = self.members.access.lock().unwrap().remove(secret); if let Some(removed) = removed { for peer in removed.peers() { - self.members.served.forget(&peer.hello.peer); + self.members.served.forget(&peer.hello.peer, None); } removed.leave("removed"); } self.members.presence(next.secret)?; *self.pairing.lock().unwrap() = Some(self.members.pair(&next)?); - let access = self.members.access.lock().unwrap(); - for device in &next.members { - if let Some(live) = access.get(&device.secret) { - live.sender().send( - kind::WELCOME, - &self.members.welcome(&next, device.secret), - None, - ); + let welcomes = { + 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(|peer| live.line(&peer.hello.peer)) + .collect::>(); + Some((self.members.welcome(&next, device.secret), lines)) + }) + .collect::>() + }; + #[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); } } (self.members.events)(); @@ -274,7 +303,7 @@ impl Members { fn open(self: &Arc, secret: [u8; 16]) -> io::Result<()> { let members = Arc::downgrade(self); - let live = Live::start( + let live = Live::start_with( Hello { serves: Some(self.sharing.lock().unwrap().share), ..self.me.clone() @@ -282,25 +311,42 @@ impl Members { &Room::Notebook(secret), self.reach, self.relay.as_deref(), - move |event| { + move |source, event| { let Some(members) = members.upgrade() else { return; }; match event { Event::Met(hello, line) => { let sharing = members.sharing.lock().unwrap(); - if !sharing.members.iter().any(|device| device.secret == secret) { - line.hang_up("removed"); - return; + let admitted = + if sharing.members.iter().any(|device| device.secret == secret) { + Ok(members + .served + .admit(source, hello, line) + .then(|| members.welcome(&sharing, secret))) + } else { + Err(()) + }; + // Browser send failures may reenter membership callbacks. + #[cfg(target_arch = "wasm32")] + drop(sharing); + match admitted { + Ok(Some(welcome)) => { + let _ = line.send(kind::WELCOME, &welcome); + } + Err(()) => { + #[cfg(not(target_arch = "wasm32"))] + drop(sharing); + line.hang_up("removed"); + } + Ok(None) => {} } - members.served.admit(hello.peer, line); - let _ = line.send(kind::WELCOME, &members.welcome(&sharing, secret)); } - Event::Left(hello) => members.served.forget(&hello.peer), + Event::Left(hello) => members.served.forget(&hello.peer, Some((source, hello))), Event::Frame { from, kind, body } if wire::KNOWN.contains(&kind) && kind > 256 && kind != kind::REPLY => { - members.served.queue(&from.peer, kind, body) + members.served.queue(source, from, kind, body) } Event::Changed => (members.events)(), _ => {} @@ -349,7 +395,7 @@ impl Members { fn pair(self: &Arc, sharing: &Sharing) -> io::Result { let members = Arc::downgrade(self); - Live::start( + Live::start_with( Hello { serves: None, ..self.me.clone() @@ -357,21 +403,41 @@ impl Members { &Room::share(&sharing.code, &sharing.password), self.reach, self.relay.as_deref(), - move |event| { + move |source, event| { let Some(members) = members.upgrade() else { return; }; match event { Event::Met(hello, line) => { - if members.sharing.lock().unwrap().approve { + let approve = members.sharing.lock().unwrap().approve; + let (approval, retired) = { let mut pending = members.pending.lock().unwrap(); - if pending.len() < 32 { - pending.insert(hello.peer, (Arc::clone(hello), line.clone())); - let _ = line.send(kind::APPROVAL, &wire::Approval::Pending); + if !source.connected(hello) { + return; + } + if approve { + if pending.len() < 32 || pending.contains_key(&hello.peer) { + ( + Some(wire::Approval::Pending), + pending.insert( + hello.peer, + ( + Arc::clone(hello), + line.clone(), + Arc::downgrade(source), + ), + ), + ) + } else { + (Some(wire::Approval::Failed), None) + } } else { - let _ = line.send(kind::APPROVAL, &wire::Approval::Failed); + (None, None) } - drop(pending); + }; + drop(retired); + if let Some(approval) = approval { + let _ = line.send(kind::APPROVAL, &approval); (members.events)(); } else if let Err(error) = members.grant(hello, line) { eprintln!("Live Share: could not admit a device: {error}"); @@ -379,7 +445,19 @@ impl Members { } } Event::Left(hello) => { - members.pending.lock().unwrap().remove(&hello.peer); + let retired = { + let mut pending = members.pending.lock().unwrap(); + if source.gone(hello) + && pending + .get(&hello.peer) + .is_some_and(|(admitted, _, _)| Arc::ptr_eq(admitted, hello)) + { + pending.remove(&hello.peer) + } else { + None + } + }; + drop(retired); (members.events)(); } Event::Changed => (members.events)(), @@ -389,3 +467,153 @@ 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" + ); + } +} diff --git a/crates/notebook/src/live/web.rs b/crates/notebook/src/live/web.rs index a0b6201e200c5d69263fb2d8543fd2e8bcbf8442..7d8eca0ceb037b16ef067bdb15349bc77e2aac7b 100644 --- a/crates/notebook/src/live/web.rs +++ b/crates/notebook/src/live/web.rs @@ -42,7 +42,8 @@ struct Shared { me: Hello, room: Room, url: String, - events: Box, + #[allow(clippy::type_complexity)] + events: Box, Event) + Send + Sync>, stopped: AtomicBool, state: Mutex, } @@ -139,9 +140,19 @@ impl Live { pub fn start( me: Hello, room: &Room, - _: Option, + reach: Option, relay: Option<&str>, events: impl Fn(Event) + Send + Sync + 'static, + ) -> io::Result { + Self::start_with(me, room, reach, relay, move |_, event| events(event)) + } + + fn start_with( + me: Hello, + room: &Room, + _: Option, + relay: Option<&str>, + events: impl Fn(&Arc, Event) + Send + Sync + 'static, ) -> io::Result { let tag = room.tag().ok_or(io::ErrorKind::InvalidInput)?; let relay = relay @@ -223,17 +234,7 @@ impl Live { Sender(Arc::downgrade(&self.0)) } pub fn peers(&self) -> Vec { - let state = self.0.state.lock().unwrap(); - let mut peers = BTreeMap::new(); - for peer in state - .members - .values() - .filter_map(|m| m.peer.as_ref()) - .chain(state.streams.values().filter_map(|s| s.peer.as_ref())) - { - peers.insert(peer.hello.peer, peer.clone()); - } - peers.into_values().collect() + self.0.peers() } pub fn line(&self, peer: &[u8; 16]) -> Option { self.0 @@ -288,11 +289,45 @@ impl Live { impl Drop for Live { fn drop(&mut self) { self.0.stopped.store(true, Ordering::Release); - close(self.0.state.lock().unwrap().socket); + let socket = self.0.state.lock().unwrap().socket; + close(socket); } } impl Shared { + fn peers(&self) -> Vec { + let state = self.state.lock().unwrap(); + let mut peers = BTreeMap::new(); + for peer in state + .members + .values() + .filter_map(|m| m.peer.as_ref()) + .chain(state.streams.values().filter_map(|s| s.peer.as_ref())) + { + peers.insert(peer.hello.peer, peer.clone()); + } + peers.into_values().collect() + } + + 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() + .is_some_and(|peer| Arc::ptr_eq(&peer.hello, hello)) + }) + } + fn connect(self: &Arc) -> io::Result<()> { let weak = Arc::downgrade(self); let message = Closure::::new(move |data: JsValue| { @@ -324,13 +359,12 @@ impl Shared { } fn disconnected(self: &Arc) { - let peers = { + let (socket, peers) = { let mut state = self.state.lock().unwrap(); let socket = std::mem::replace(&mut state.socket, 0); if socket == 0 { return; } - close(socket); let peers: Vec<_> = state .streams .values() @@ -343,12 +377,13 @@ impl Shared { state.members.clear(); state.sealer = None; state.relayed = Relayed::Unreachable(Trouble::Other); - peers + (socket, peers) }; + close(socket); for hello in peers { - (self.events)(Event::Left(&hello)); + (self.events)(self, Event::Left(&hello)); } - (self.events)(Event::Changed); + (self.events)(self, Event::Changed); let weak = Arc::downgrade(self); let _ = crate::task::spawn("live reconnect", move || async move { let (_, wait) = crate::task::channel(); @@ -464,13 +499,13 @@ impl Shared { }) }; if let Some(hello) = hello { - (self.events)(Event::Left(&hello)); + (self.events)(self, Event::Left(&hello)); } } ::relay::Notice::Burned => self.state.lock().unwrap().burned = true, _ => {} } - (self.events)(Event::Changed); + (self.events)(self, Event::Changed); return Ok(()); } let bytes = js_sys::Uint8Array::new(&data).to_vec(); @@ -523,7 +558,7 @@ impl Shared { if serves { self.meet(slot)?; } - (self.events)(Event::Changed); + (self.events)(self, Event::Changed); } kind::PRESENCE => { let presence = minicbor::decode(&body).map_err(|_| io::ErrorKind::InvalidData)?; @@ -537,15 +572,18 @@ impl Shared { { peer.presence = Some(presence); } - (self.events)(Event::Changed); + (self.events)(self, Event::Changed); } kind if kind < 256 => { if let Some(hello) = hello { - (self.events)(Event::Frame { - from: &hello, - kind, - body: &body, - }); + (self.events)( + self, + Event::Frame { + from: &hello, + kind, + body: &body, + }, + ); } } _ => {} @@ -568,7 +606,9 @@ impl Shared { loop { let event = { let mut state = self.state.lock().unwrap(); - let stream = state.streams.get_mut(&slot).unwrap(); + let Some(stream) = state.streams.get_mut(&slot) else { + break; + }; let Some(length) = stream.bytes.get(..4) else { break; }; @@ -613,12 +653,17 @@ impl Shared { if let Some((line, hello, kind, body)) = event { match hello { None => line.send(kind::HELLO, &self.me)?, - Some(hello) if kind == kind::HELLO => (self.events)(Event::Met(&hello, &line)), - Some(hello) => (self.events)(Event::Frame { - from: &hello, - kind, - body: &body, - }), + Some(hello) if kind == kind::HELLO => { + (self.events)(self, Event::Met(&hello, &line)) + } + Some(hello) => (self.events)( + self, + Event::Frame { + from: &hello, + kind, + body: &body, + }, + ), } } } diff --git a/crates/notebook/tests/live_membership.rs b/crates/notebook/tests/live_membership.rs index 0f3141bc2bf6ccf966396d734c592c8d4884294b..8482e53ae6beeea0f4727132e892d9ca63e12abd 100644 --- a/crates/notebook/tests/live_membership.rs +++ b/crates/notebook/tests/live_membership.rs @@ -104,6 +104,9 @@ fn removal_retires_one_credential_and_other_devices_reconnect_after_a_restart() &directory.path().join("alice"), ); let (bob, _) = guest("Bob", &original_code, &url, &directory.path().join("bob")); + until("the initial presence room never met both devices", || { + host.guests().len() == 2 + }); let before = host.sharing(); let alice_key = before .members @@ -122,7 +125,8 @@ fn removal_retires_one_credential_and_other_devices_reconnect_after_a_restart() bob.ended() == Some(share::Ended::Removed) && bob.host().is_none() }); until("presence did not move to the new room", || { - host.guests().len() == 1 + let guests = host.guests(); + guests.len() == 1 && guests[0].hello.name == "Alice" }); let current = host.sharing(); assert_ne!(before.secret, current.secret);