From a46bcc611e2e0f0799e9271027594e97e0df9e92 Mon Sep 17 00:00:00 2001 From: clover caruso Date: Mon, 12 Oct 2026 22:37:02 -0700 Subject: [PATCH] fix(live): fence handover control frames to current streams Route credential rotation through authenticated access streams and serialize native welcome snapshots with their enqueue. Release browser membership locks before fallible sends and stop a superseded rotation batch. Preserve stream incarnation checks and terminal retirement. Fixes #141 Assisted-by: gpt-6.1-sol --- crates/notebook/src/live.rs | 79 +- crates/notebook/src/live/relay.rs | 29 +- crates/notebook/src/live/share.rs | 432 +++++++---- crates/notebook/src/live/share/batch.rs | 4 + .../notebook/src/live/share/handover_tests.rs | 721 ++++++++++++++++++ crates/notebook/src/live/share/membership.rs | 302 +++++++- crates/notebook/src/live/web.rs | 115 ++- crates/notebook/tests/live_membership.rs | 6 +- 8 files changed, 1443 insertions(+), 245 deletions(-) create mode 100644 crates/notebook/src/live/share/handover_tests.rs 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); -- 2.54.0