diff --git a/crates/notebook/src/live/share.rs b/crates/notebook/src/live/share.rs index 7b966c3e5e9f0fef0a429111ab28d411a0dba72c..5f2dc1103bd0fd976b6b0521565b0b8ca451427d 100644 --- a/crates/notebook/src/live/share.rs +++ b/crates/notebook/src/live/share.rs @@ -1003,7 +1003,7 @@ struct Inner { pending: Mutex>>, next: AtomicU64, /// Where the host's reports of changed files go, while a background watches. - watch: Mutex>, + watch: Mutex>>, /// Why this device can no longer reach the share. ended: Mutex>, /// Bytes of chunks asked for and not yet given. @@ -1132,7 +1132,7 @@ impl Guest { if self.host().is_none() { return Err(self.offline()); } - *self.inner.watch.lock().unwrap() = Some(reports); + *self.inner.watch.lock().unwrap() = Some(Arc::new(reports)); Ok(()) } @@ -1610,7 +1610,8 @@ impl Inner { kind::TOUCHED if serves(from) => { if let Ok(touched) = minicbor::decode::(body) { self.touched(&touched); - if let Some(reports) = &*self.watch.lock().unwrap() { + let reports = self.watch.lock().unwrap().clone(); + if let Some(reports) = reports { reports.touched(&touched.paths); } } @@ -1639,6 +1640,7 @@ impl Inner { return; } let inner = Arc::downgrade(self); + let connected = Mutex::new(None); let live = Live::start( self.me.clone(), &Room::Notebook(secret), @@ -1646,6 +1648,35 @@ impl Inner { 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()]); + } + } + } if matches!( event, Event::Frame { @@ -1663,11 +1694,13 @@ impl Inner { 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); - if let Some(reports) = &*self.watch.lock().unwrap() { + let reports = self.watch.lock().unwrap().clone(); + if let Some(reports) = reports { reports.touched(&paths); } } diff --git a/crates/notebook/tests/live_share.rs b/crates/notebook/tests/live_share.rs index 885534822a2b6c9ed79d1bbd1c32915450098769..6ec3cd9e0bf1e8684dc56415882b7af92df585f6 100644 --- a/crates/notebook/tests/live_share.rs +++ b/crates/notebook/tests/live_share.rs @@ -53,6 +53,147 @@ fn a_guest_edits_the_host_s_notebook() { }); } +#[test] +fn presence_reconnect_reconciles_missed_changes_without_reconnecting_access() { + use notebook::{ + Remote, + live::{ + Caret, Presence, Spot, + share::{Guest, HostedRemote}, + }, + session::Background, + }; + use onestore::Stamp; + use std::sync::atomic::{AtomicUsize, Ordering}; + + let directory = tempfile::tempdir().unwrap(); + let folder = notebook(directory.path()); + let url = relay(Default::default()); + let sharing = Sharing::new("").unwrap(); + let host = host(&folder, &directory.path().join("host"), &sharing, &url); + let welcome = share::join(hello("Grace"), &code(&host), "", None, Some(&url)).unwrap(); + let relay = PresenceRelay::new(&url, &welcome.room); + let guest = Guest::start( + hello("Grace"), + welcome.share, + welcome.secret, + None, + Some(&relay.url), + || {}, + ) + .unwrap(); + until("the access host was never met", || guest.host().is_some()); + let host_id = guest.host().unwrap().peer; + until("the presence host was never met", || { + guest.peers().iter().any(|peer| peer.hello.peer == host_id) + }); + let mut notebook = + Notebook::open_hosted(Arc::clone(&guest), directory.path().join("grace")).unwrap(); + let section = open(¬ebook, &guest, "Garden.one", None); + let background = Background::hosted(Arc::clone(&guest), || {}).unwrap(); + background.watch(notebook.replicas()); + background.hold("Garden.one", §ion); + let storage = notebook.into_storage(); + until("the section never synced", || { + section.sync_status().unwrap().synced.is_some() + }); + let touched = Arc::new(AtomicUsize::new(0)); + let heard = Arc::clone(&touched); + let observed = Arc::clone(&guest); + background.on_touched(Some(Box::new(move |_| { + observed.set_presence(Presence::default()); + HostedRemote::new(&observed, "Garden.one").stamp().unwrap(); + heard.fetch_add(1, Ordering::SeqCst); + }))); + let file = folder.join("Garden.one"); + let image = std::fs::read(&file).unwrap(); + let (space, text, _) = server::text(&image); + let stamp = Stamp::of(&image).unwrap(); + until("the initial TOUCHED never arrived", || { + host.touched(&["Garden.one".into()]); + touched.load(Ordering::SeqCst) > 0 + }); + let mut remote = HostedRemote::new(&guest, "Garden.one"); + until("TOUCHED did not cache the current stamp", || { + let before = relay.state.lock().unwrap().access_bytes; + assert_eq!( + HostedRemote::new(&guest, "Garden.one").stamp().unwrap(), + stamp + ); + relay.state.lock().unwrap().access_bytes == before + }); + let before = relay.state.lock().unwrap().access_bytes; + for at in 1..=8 { + let spot = Spot { + text: text.into(), + offset: at, + }; + let presence = Presence { + page: Some(space.into()), + caret: Some(Caret { + anchor: spot, + focus: spot, + }), + ..Presence::default() + }; + host.set_presence(presence.clone()); + until("ordinary presence never reached the guest", || { + guest + .peers() + .iter() + .any(|peer| peer.hello.peer == host_id && peer.presence.as_ref() == Some(&presence)) + }); + assert_eq!(remote.stamp().unwrap(), stamp); + } + assert_eq!( + relay.state.lock().unwrap().access_bytes, + before, + "typing caused remote stamp or image requests" + ); + + relay.disconnect_presence(); + until("the presence host never left", || { + !guest.peers().iter().any(|peer| peer.hello.peer == host_id) + }); + assert_eq!(guest.host().unwrap().peer, host_id); + let access_connections = relay.state.lock().unwrap().access_connections; + assert_eq!(access_connections, 1); + std::fs::write(folder.join("reachable.bin"), b"still connected").unwrap(); + assert_eq!( + storage.read_file("reachable.bin", 1024).unwrap(), + b"still connected" + ); + let missed = server::typed(&image, space, text, 0..8, "Missed"); + std::fs::write(&file, &missed).unwrap(); + let before = touched.load(Ordering::SeqCst); + host.touched(&["Garden.one".into()]); + std::thread::sleep(std::time::Duration::from_millis(200)); + assert_eq!( + touched.load(Ordering::SeqCst), + before, + "the isolated guest received TOUCHED" + ); + assert!( + server::page_texts(§ion.page(space).unwrap()).contains(&"Original text".to_owned()) + ); + + relay.state.lock().unwrap().blocked = false; + until("the presence host never returned", || { + guest.peers().iter().any(|peer| peer.hello.peer == host_id) + }); + until( + "the missed change never reconciled after presence returned", + || server::page_texts(§ion.page(space).unwrap()).contains(&"Missed text".to_owned()), + ); + assert_eq!(guest.host().unwrap().peer, host_id); + assert_eq!( + relay.state.lock().unwrap().access_connections, + access_connections + ); + assert_eq!(section.replica().snapshot().unwrap(), missed); + background.stop(); +} + /// While the host is away a guest's edits wait in its replica, the notebook opens from its /// last listing, and once the host is back the edits reach its file. #[test] diff --git a/crates/notebook/tests/support/live.rs b/crates/notebook/tests/support/live.rs index 96af2f076a6314ce6406dd63a9e4218e6189cdb2..b080fdbb52bb37bff28955b24d1a4965e5bb98d0 100644 --- a/crates/notebook/tests/support/live.rs +++ b/crates/notebook/tests/support/live.rs @@ -156,3 +156,137 @@ pub fn published(section: &Section, id: u64) { ) }); } + +pub struct PresenceRelay { + pub url: String, + pub state: Arc>, + thread: Option>, +} + +#[derive(Default)] +pub struct RelayState { + pub blocked: bool, + pub access_connections: usize, + pub access_bytes: usize, + sockets: Vec<(bool, std::net::TcpStream)>, + stopped: bool, +} + +impl PresenceRelay { + pub fn new(upstream: &str, secret: &[u8; 16]) -> Self { + use sha2::{Digest, Sha256}; + use std::{ + io::{self, Read, Write}, + net::{Shutdown, TcpStream}, + sync::Mutex, + }; + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + let upstream = upstream.strip_prefix("ws://").unwrap().to_owned(); + let tag: String = Sha256::digest([&b"Snowbound room "[..], secret].concat())[..8] + .iter() + .map(|byte| format!("{byte:02x}")) + .collect(); + let path = format!("/v1/room/{tag}"); + let state = Arc::new(Mutex::new(RelayState::default())); + let serving = Arc::clone(&state); + let thread = thread::spawn(move || { + let mut workers = Vec::new(); + for client in listener.incoming().flatten() { + if serving.lock().unwrap().stopped { + break; + } + let state = Arc::clone(&serving); + let upstream = upstream.clone(); + let path = path.clone(); + workers.push(thread::spawn(move || { + let _ = (|| -> io::Result<()> { + let mut client = client; + client.set_read_timeout(Some(Duration::from_secs(5)))?; + let head = relay::ws::head(&mut client)?; + let presence = head + .split(' ') + .nth(1) + .map(|target| target.split('?').next().unwrap()) + == Some(path.as_str()); + let mut upstream = TcpStream::connect(upstream)?; + { + let mut state = state.lock().unwrap(); + if state.stopped { + return Ok(()); + } + if presence && state.blocked { + client.write_all(b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n")?; + return Ok(()); + } + state.sockets.push((presence, client.try_clone()?)); + state.sockets.push((presence, upstream.try_clone()?)); + if !presence { + state.access_connections += 1; + } + } + upstream.write_all(head.as_bytes())?; + client.set_read_timeout(None)?; + let (mut from, mut to) = (client.try_clone()?, upstream.try_clone()?); + let counted = Arc::clone(&state); + let requests = thread::spawn(move || { + let mut bytes = [0; 16 << 10]; + while let Ok(length) = from.read(&mut bytes) { + if length == 0 { + break; + } + if !presence { + counted.lock().unwrap().access_bytes += length; + } + if to.write_all(&bytes[..length]).is_err() { + break; + } + } + let _ = to.shutdown(Shutdown::Both); + }); + let _ = io::copy(&mut upstream, &mut client); + let _ = client.shutdown(Shutdown::Both); + let _ = requests.join(); + Ok(()) + })(); + })); + } + for worker in workers { + let _ = worker.join(); + } + }); + Self { + url, + state, + thread: Some(thread), + } + } + + pub fn disconnect_presence(&self) { + let mut state = self.state.lock().unwrap(); + state.blocked = true; + state.sockets.retain(|(presence, socket)| { + if *presence { + let _ = socket.shutdown(std::net::Shutdown::Both); + } + !*presence + }); + } +} + +impl Drop for PresenceRelay { + fn drop(&mut self) { + { + let mut state = self.state.lock().unwrap(); + state.stopped = true; + for (_, socket) in &state.sockets { + let _ = socket.shutdown(std::net::Shutdown::Both); + } + } + let _ = std::net::TcpStream::connect(self.url.strip_prefix("ws://").unwrap()); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +}