diff --git a/crates/notebook/README.md b/crates/notebook/README.md index 3d653b8c6bbe37eb6d37892d9ecefde58e226b12..29451bfc0e3c7bcba279efa7fb5fa7d40b2e6eb7 100644 --- a/crates/notebook/README.md +++ b/crates/notebook/README.md @@ -52,9 +52,12 @@ Share one `Replica` between threads. Keep the cache on a local filesystem: the connection holds exclusive ownership between transactions, and a second open fails busy. No network wait occurs in a local edit. An edit a session submitted that the cache cannot write for now (a full disk, an I/O error) stays applied in memory and is written -again, in order, each second; meanwhile `written` returns `WouldBlock`. After a commit -fails, a fresh connection decides whether it landed. An edit the cache could never write -is refused alone, and the section thread rereads the queue. +again, in order, each second; meanwhile `written` returns `WouldBlock`. A replica dropped +meanwhile keeps its cache open and its edits retrying until they are written or the cache is +deleted, and only then releases it; `notebook::unwritten()` counts the edits waiting in the +process, for a host to ask before it quits. After a commit fails, a fresh connection +decides whether it landed. An edit the cache could never write is refused alone, and the +section thread rereads the queue. ## Sessions diff --git a/crates/notebook/src/lib.rs b/crates/notebook/src/lib.rs index 450ca9d67399a93af8a62bf9f2bd397b23d22232..9e34335bd567e3f1569dfaa609c610405a9b8a39 100644 --- a/crates/notebook/src/lib.rs +++ b/crates/notebook/src/lib.rs @@ -131,6 +131,15 @@ pub struct Replica { root: ExGuid, } +/// Edits that sections of this process hold in memory because their cache refused them for now, +/// a dropped replica's included, which its section thread keeps writing. +static UNWRITTEN: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); + +/// How many edits wait in memory for a cache to write them; quitting loses them. +pub fn unwritten() -> usize { + UNWRITTEN.load(std::sync::atomic::Ordering::Relaxed) +} + /// Runs once the replica is released and its cache closed, as `Background::hold` asks. #[derive(Default)] struct Released(Mutex>>); diff --git a/crates/notebook/src/working.rs b/crates/notebook/src/working.rs index 63d4fa2283f9022000485207e2419724ac70ea63..99fa5e52ea9e671447ee0a7d0d209834665fe3db 100644 --- a/crates/notebook/src/working.rs +++ b/crates/notebook/src/working.rs @@ -16,7 +16,11 @@ use std::{ collections::{BTreeMap, BTreeSet, VecDeque}, io, path::PathBuf, - sync::{Arc, Mutex, Weak, mpsc}, + sync::{ + Arc, Mutex, Weak, + atomic::{AtomicUsize, Ordering}, + mpsc, + }, }; pub(crate) type Reply = Box) + Send>; @@ -121,11 +125,25 @@ pub(crate) struct Thread { threads: Mutex>>, /// A password-protected section's key, under which its queue is sealed too. pub(crate) key: Option, + /// The edits the cache last refused for now, as `crate::unwritten` counts them. + waiting: AtomicUsize, /// Last, so the cache closes before background synchronization takes it over. pub(crate) released: crate::Released, } +impl Drop for Thread { + fn drop(&mut self) { + self.wait(0); + } +} + impl Thread { + fn wait(&self, edits: usize) { + let was = self.waiting.swap(edits, Ordering::Relaxed); + crate::UNWRITTEN.fetch_add(edits, Ordering::Relaxed); + crate::UNWRITTEN.fetch_sub(was, Ordering::Relaxed); + } + pub(crate) fn send(&self, request: Request) -> Result<()> { self.sender .lock() @@ -341,6 +359,7 @@ pub(crate) fn spawn(connection: Connection, key: Option) -> Result<(Arc, requests: mpsc::Receiver) { +fn fail( + shared: &Thread, + error: crate::Error, + backlog: VecDeque, + requests: mpsc::Receiver, +) { + // The edits that waited are in the backlog, refused with the rest. + shared.wait(0); let message = error.to_string(); for request in backlog.into_iter().chain(requests.iter()) { request.fail(io::ErrorKind::Other, &message); @@ -449,7 +475,7 @@ fn build(shared: Arc, job: Job, signal: mpsc::Sender) { }; let working = match opened { Ok(working) => working, - Err(error) => return fail(error, backlog, requests), + Err(error) => return fail(&shared, error, backlog, requests), }; match crate::task::complete(working.serve(&shared, &requests, &mut backlog)) { Next::Stop => {} @@ -479,6 +505,34 @@ fn inline(shared: &Thread, job: Job) -> bool { true } +/// Writes again each second the edits a dropped replica left waiting, until none do or their +/// cache is gone, keeping the cache open and `Released` from firing meanwhile. +#[cfg(not(target_arch = "wasm32"))] +fn orphan(shared: &Arc, unwritten: Vec) { + let shared = Arc::clone(shared); + let path = lock(&shared.connection) + .ok() + .and_then(|connection| connection.path().map(PathBuf::from)); + let spawned = std::thread::Builder::new() + .name("onestore-unwritten".into()) + .spawn(move || { + let arena = Arena::default(); + let Ok(mut working) = Working::open(&arena, &shared.connection, shared.key.as_ref()) + else { + return; + }; + working.unwritten = unwritten; + // A cache deleted, as with its notebook, takes no edits. + while !working.unwritten.is_empty() && path.as_ref().is_some_and(|path| path.exists()) { + std::thread::sleep(std::time::Duration::from_secs(1)); + let _ = working.flush(&shared); + } + }); + if let Err(error) = spawned { + eprintln!("Edits waiting for the cache are lost: {error}"); + } +} + /// An edit to apply, with who waits for it. pub(crate) struct Accepted { pub author: String, @@ -532,6 +586,10 @@ impl<'a> Working<'a> { Err(mpsc::RecvTimeoutError::Timeout) => write = true, Err(mpsc::RecvTimeoutError::Disconnected) => { let _ = self.flush(shared); + #[cfg(not(target_arch = "wasm32"))] + if !self.unwritten.is_empty() { + orphan(shared, std::mem::take(&mut self.unwritten)); + } return Next::Stop; } } @@ -810,6 +868,7 @@ impl<'a> Working<'a> { }) { (edit.reply)(Err(io::Error::other(message.clone()).into())); } + shared.wait(self.unwritten.len()); if let Some(notice) = self .unwritten .first() @@ -845,6 +904,7 @@ impl<'a> Working<'a> { } crate::edited(&shared.worker); } + shared.wait(0); Ok(()) } diff --git a/crates/snowbound/src/library.rs b/crates/snowbound/src/library.rs index a96c76975d61defde51827582b344d572b24f1eb..0f0d8c31a2bd6a59abdb6c5141b551a78970a931 100644 --- a/crates/snowbound/src/library.rs +++ b/crates/snowbound/src/library.rs @@ -1924,6 +1924,82 @@ mod tests { notebook::fs::remove_dir_all(&root).unwrap(); } + /// An edit the cache can't write yet outlives switching away from its section, counts + /// toward quitting, and is in the cache once the cache takes it, as a cold open reads. + #[test] + fn an_edit_waiting_for_the_cache_outlives_its_section() { + use onestore::op::{Edit, Op, SectionOp}; + + let root = std::env::temp_dir().join(format!("snowbound-waiting-{}", std::process::id())); + let _ = notebook::fs::remove_dir_all(&root); + let folder = root.join("Personal"); + notebook::fs::create_dir_all(&folder).unwrap(); + let sample = + Path::new(env!("CARGO_MANIFEST_DIR")).join("../../docs/sample-notebook/Personal"); + for entry in notebook::fs::read_dir(sample).unwrap() { + let entry = entry.unwrap(); + notebook::fs::copy(entry.path(), folder.join(entry.file_name())).unwrap(); + } + let library = Library::notebook(folder.to_str().unwrap(), &root.join("cache")); + let paths: Vec = library.tabs("").into_iter().map(|tab| tab.path).collect(); + let first = &paths[0]; + let notebook = library.notebook.as_ref().unwrap().as_ref().unwrap(); + let cache = notebook.replica_path(first).unwrap(); + library.open(first, || {}).unwrap().close().unwrap(); + // The cache takes no writes from here until the test lets it. Holding the lock keeps + // every other test's cache writing, so the process counts this test's edit alone. + crate::vfs::install(); + let _exclusive = + (crate::vfs::EXCLUSIVE.lock()).unwrap_or_else(std::sync::PoisonError::into_inner); + let deadline = Instant::now() + Duration::from_secs(10); + while notebook::unwritten() > 0 { + assert!(Instant::now() < deadline, "another test's edits still wait"); + std::thread::sleep(Duration::from_millis(10)); + } + *crate::vfs::FAILING.lock().unwrap() = Some(cache.canonicalize().unwrap()); + let section = library.open(first, || {}).unwrap(); + let creation = onestore::PageCreation::new(None, Some("Waiting"), "Fixture").unwrap(); + let ops = vec![Op::Section(SectionOp::Create(creation))]; + let at = crate::filetime(); + section.apply("Fixture", Edit { at, ops }).unwrap(); + let deadline = Instant::now() + Duration::from_secs(10); + while notebook::unwritten() == 0 { + assert!(Instant::now() < deadline, "the cache took the edit"); + std::thread::sleep(Duration::from_millis(10)); + } + + // Switching through every other section closes the first and drops its replica. + let replica = Arc::downgrade(section.replica()); + library.keep(first, section); + for path in &paths[1..] { + library.keep(path, library.open(path, || {}).unwrap()); + } + while replica.strong_count() > 0 { + assert!(Instant::now() < deadline, "the section stayed open"); + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!(notebook::unwritten(), 1); + + *crate::vfs::FAILING.lock().unwrap() = None; + while notebook::unwritten() > 0 { + assert!(Instant::now() < deadline, "the edit was not written"); + std::thread::sleep(Duration::from_millis(10)); + } + let cold = loop { + match notebook::Replica::open(&cache) { + Ok(cold) => break cold, + Err(error) if error.busy() && Instant::now() < deadline => { + std::thread::sleep(Duration::from_millis(10)); + } + Err(error) => panic!("{error}"), + } + }; + let pages = cold.pages().unwrap(); + assert!(pages.iter().any(|(_, title, _)| title == "Waiting")); + drop((cold, library)); + notebook::fs::remove_dir_all(&root).unwrap(); + } + #[cfg(unix)] #[test] fn within_mount_resolves_symlinks() { diff --git a/crates/snowbound/src/main.rs b/crates/snowbound/src/main.rs index e08a1550ae6ffdb4d2bab035bced1a2ce62914a1..763bca69935c7e3482200f8be04b2a11faf3ac27 100644 --- a/crates/snowbound/src/main.rs +++ b/crates/snowbound/src/main.rs @@ -6544,35 +6544,28 @@ impl State { } #[cfg(not(target_arch = "wasm32"))] -impl App { - fn close(&self, event_loop: &ActiveEventLoop) { - let Some(state) = self.state.as_ref().filter(|state| { - let editor = &state.view.editor; - let unchanged = editor.caret_outline().is_none_or(TextOutline::is_empty) - && state.initial_date == editor.date().map(|date| date.timestamp()) - && state - .initial_layouts - .iter() - .map(|(id, layout)| (*id, layout)) - .eq(editor.object_layouts()) - && state - .initial - .iter() - .map(|(id, document)| (id, document)) - .eq(editor - .outlines() - .iter() - .map(|outline| (&outline.id, outline.document()))); - state.session.is_none() && !unchanged - }) else { - return event_loop.exit(); - }; +impl State { + /// Quits, first asking whether to discard a temporary page holding edits. + fn quit(&self) { + let editor = &self.view.editor; + let unchanged = editor.caret_outline().is_none_or(TextOutline::is_empty) + && self.initial_date == editor.date().map(|date| date.timestamp()) + && (self.initial_layouts.iter()) + .map(|(id, layout)| (*id, layout)) + .eq(editor.object_layouts()) + && (self.initial.iter()) + .map(|(id, document)| (id, document)) + .eq((editor.outlines().iter()).map(|outline| (&outline.id, outline.document()))); + if self.session.is_some() || unchanged { + let _ = self.proxy.send_event(UserEvent::Exit); + return; + } crate::confirm( "Discard this page?", "This temporary page has no saved copy. Closing it will discard your edits.", "Keep Editing", "Discard Changes", - state.reply(|state, ()| { + self.reply(|state, ()| { let _ = state.proxy.send_event(UserEvent::Exit); Ok(()) }), @@ -6580,6 +6573,31 @@ impl App { } } +#[cfg(not(target_arch = "wasm32"))] +impl App { + fn close(&self, event_loop: &ActiveEventLoop) { + let Some(state) = &self.state else { + return event_loop.exit(); + }; + let changes = match notebook::unwritten() { + 0 => return state.quit(), + 1 => "1 unsaved change".to_owned(), + count => format!("{count} unsaved changes"), + }; + crate::confirm( + &format!("Quit with {changes}?"), + "Snowbound keeps trying to save while it’s open. Quitting now loses what hasn’t \ + been saved.", + "Keep Snowbound Open", + "Quit Anyway", + state.reply(|state, ()| { + state.quit(); + Ok(()) + }), + ); + } +} + /// How long quitting waits for the open section's edits to reach its file and for every /// section to close; edits left stay in the replica and publish at the next launch. #[cfg(not(target_arch = "wasm32"))]