From 5d52a7a3ed5060df052284db1bd6cd66fe57db96 Mon Sep 17 00:00:00 2001 From: clover caruso Date: Thu, 15 Oct 2026 06:17:07 -0700 Subject: [PATCH] fix(desktop): keep edits the cache could not write past closing their section A section whose edits the local cache could not write yet, through a full disk, an I/O error or a lock, lost them when its replica was dropped: switching through more sections than desktop keeps open, closing a notebook, or locking a protected section. A replica dropped while edits wait now leaves its section thread writing them each second, with the cache held open, until the cache takes them or is deleted with its notebook; only then is the cache released and the background takes the section over as usual. Reopening that section meanwhile waits as for any held cache. Quitting while edits wait anywhere in the app asks first, counting them with notebook::unwritten(). Keep Snowbound Open is the default; Quit Anyway goes on to the usual question about a temporary page. Fixes #147 Assisted-by: claude-opus-5.5 --- crates/notebook/README.md | 9 ++-- crates/notebook/src/lib.rs | 9 ++++ crates/notebook/src/working.rs | 68 +++++++++++++++++++++++++++-- crates/snowbound/src/library.rs | 76 +++++++++++++++++++++++++++++++++ crates/snowbound/src/main.rs | 66 +++++++++++++++++----------- 5 files changed, 197 insertions(+), 31 deletions(-) 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"))] -- 2.54.0