From 1f48c20a72a50fe2b69f1cde84c84b025e52973c Mon Sep 17 00:00:00 2001 From: clover caruso Date: Mon, 5 Oct 2026 22:30:35 -0700 Subject: [PATCH] fix: reconcile changes after Live Share presence returns Invalidate cached host stamps across presence connectivity transitions and reconcile missed edits when presence returns while preserving the access session. Release callback locks before notifying listeners. fixes #96 Assisted-by: gpt-6.1-sol --- crates/notebook/src/live/share.rs | 41 +++++++- crates/notebook/tests/live_share.rs | 141 ++++++++++++++++++++++++++ crates/notebook/tests/support/live.rs | 134 ++++++++++++++++++++++++ 3 files changed, 312 insertions(+), 4 deletions(-) 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(); + } + } +} -- 2.54.0