authorclover caruso <git@paperclover.net> 2026-10-15 06:17:07-07:00
committerclover caruso <git@paperclover.net> 2026-10-15 15:49:32-07:00
log5d52a7a3ed5060df052284db1bd6cd66fe57db96
treeb5440d21385aa25bc012579d113779fcde6cace9
parentc345a48e944a431673a0e8b9d4a59e93dc507a58
signature Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU (~clover)

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

5 files changed, 197 insertions(+), 31 deletions(-)

crates/notebook/README.md+6-3
...@@ -52,9 +52,12 @@ Share one `Replica` between threads. Keep the cache on a local filesystem: the...@@ -52,9 +52,12 @@ Share one `Replica` between threads. Keep the cache on a local filesystem: the
52connection holds exclusive ownership between transactions, and a second open fails52connection holds exclusive ownership between transactions, and a second open fails
53busy. No network wait occurs in a local edit. An edit a session submitted that the cache53busy. No network wait occurs in a local edit. An edit a session submitted that the cache
54cannot write for now (a full disk, an I/O error) stays applied in memory and is written54cannot write for now (a full disk, an I/O error) stays applied in memory and is written
55again, in order, each second; meanwhile `written` returns `WouldBlock`. After a commit55again, in order, each second; meanwhile `written` returns `WouldBlock`. A replica dropped
56fails, a fresh connection decides whether it landed. An edit the cache could never write56meanwhile keeps its cache open and its edits retrying until they are written or the cache is
57is refused alone, and the section thread rereads the queue.57deleted, and only then releases it; `notebook::unwritten()` counts the edits waiting in the
58process, for a host to ask before it quits. After a commit fails, a fresh connection
59decides whether it landed. An edit the cache could never write is refused alone, and the
60section thread rereads the queue.
5861
59## Sessions62## Sessions
6063
crates/notebook/src/lib.rs+9
...@@ -131,6 +131,15 @@ pub struct Replica {...@@ -131,6 +131,15 @@ pub struct Replica {
131 root: ExGuid,131 root: ExGuid,
132}132}
133133
134/// Edits that sections of this process hold in memory because their cache refused them for now,
135/// a dropped replica's included, which its section thread keeps writing.
136static UNWRITTEN: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
137
138/// How many edits wait in memory for a cache to write them; quitting loses them.
139pub fn unwritten() -> usize {
140 UNWRITTEN.load(std::sync::atomic::Ordering::Relaxed)
141}
142
134/// Runs once the replica is released and its cache closed, as `Background::hold` asks.143/// Runs once the replica is released and its cache closed, as `Background::hold` asks.
135#[derive(Default)]144#[derive(Default)]
136struct Released(Mutex<Option<Box<dyn FnOnce() + Send>>>);145struct Released(Mutex<Option<Box<dyn FnOnce() + Send>>>);
crates/notebook/src/working.rs+64-4
...@@ -16,7 +16,11 @@ use std::{...@@ -16,7 +16,11 @@ use std::{
16 collections::{BTreeMap, BTreeSet, VecDeque},16 collections::{BTreeMap, BTreeSet, VecDeque},
17 io,17 io,
18 path::PathBuf,18 path::PathBuf,
19 sync::{Arc, Mutex, Weak, mpsc},19 sync::{
20 Arc, Mutex, Weak,
21 atomic::{AtomicUsize, Ordering},
22 mpsc,
23 },
20};24};
2125
22pub(crate) type Reply<T> = Box<dyn FnOnce(Result<T>) + Send>;26pub(crate) type Reply<T> = Box<dyn FnOnce(Result<T>) + Send>;
...@@ -121,11 +125,25 @@ pub(crate) struct Thread {...@@ -121,11 +125,25 @@ pub(crate) struct Thread {
121 threads: Mutex<Vec<std::thread::JoinHandle<()>>>,125 threads: Mutex<Vec<std::thread::JoinHandle<()>>>,
122 /// A password-protected section's key, under which its queue is sealed too.126 /// A password-protected section's key, under which its queue is sealed too.
123 pub(crate) key: Option<Key>,127 pub(crate) key: Option<Key>,
128 /// The edits the cache last refused for now, as `crate::unwritten` counts them.
129 waiting: AtomicUsize,
124 /// Last, so the cache closes before background synchronization takes it over.130 /// Last, so the cache closes before background synchronization takes it over.
125 pub(crate) released: crate::Released,131 pub(crate) released: crate::Released,
126}132}
127133
134impl Drop for Thread {
135 fn drop(&mut self) {
136 self.wait(0);
137 }
138}
139
128impl Thread {140impl Thread {
141 fn wait(&self, edits: usize) {
142 let was = self.waiting.swap(edits, Ordering::Relaxed);
143 crate::UNWRITTEN.fetch_add(edits, Ordering::Relaxed);
144 crate::UNWRITTEN.fetch_sub(was, Ordering::Relaxed);
145 }
146
129 pub(crate) fn send(&self, request: Request) -> Result<()> {147 pub(crate) fn send(&self, request: Request) -> Result<()> {
130 self.sender148 self.sender
131 .lock()149 .lock()
...@@ -341,6 +359,7 @@ pub(crate) fn spawn(connection: Connection, key: Option<Key>) -> Result<(Arc<Thr...@@ -341,6 +359,7 @@ pub(crate) fn spawn(connection: Connection, key: Option<Key>) -> Result<(Arc<Thr
341 #[cfg(not(target_arch = "wasm32"))]359 #[cfg(not(target_arch = "wasm32"))]
342 threads: Mutex::new(Vec::new()),360 threads: Mutex::new(Vec::new()),
343 key,361 key,
362 waiting: AtomicUsize::new(0),
344 released: crate::Released::default(),363 released: crate::Released::default(),
345 });364 });
346 #[cfg(target_arch = "wasm32")]365 #[cfg(target_arch = "wasm32")]
...@@ -380,7 +399,7 @@ async fn run(...@@ -380,7 +399,7 @@ async fn run(
380 if let Some(ready) = ready.take() {399 if let Some(ready) = ready.take() {
381 let _ = ready.send(Err(error));400 let _ = ready.send(Err(error));
382 } else {401 } else {
383 fail(error, backlog, requests);402 fail(&shared, error, backlog, requests);
384 }403 }
385 return;404 return;
386 }405 }
...@@ -400,7 +419,14 @@ async fn run(...@@ -400,7 +419,14 @@ async fn run(
400}419}
401420
402/// Answers every request with `error` until the replica drops.421/// Answers every request with `error` until the replica drops.
403fn fail(error: crate::Error, backlog: VecDeque<Request>, requests: mpsc::Receiver<Request>) {422fn fail(
423 shared: &Thread,
424 error: crate::Error,
425 backlog: VecDeque<Request>,
426 requests: mpsc::Receiver<Request>,
427) {
428 // The edits that waited are in the backlog, refused with the rest.
429 shared.wait(0);
404 let message = error.to_string();430 let message = error.to_string();
405 for request in backlog.into_iter().chain(requests.iter()) {431 for request in backlog.into_iter().chain(requests.iter()) {
406 request.fail(io::ErrorKind::Other, &message);432 request.fail(io::ErrorKind::Other, &message);
...@@ -449,7 +475,7 @@ fn build(shared: Arc<Thread>, job: Job, signal: mpsc::Sender<Request>) {...@@ -449,7 +475,7 @@ fn build(shared: Arc<Thread>, job: Job, signal: mpsc::Sender<Request>) {
449 };475 };
450 let working = match opened {476 let working = match opened {
451 Ok(working) => working,477 Ok(working) => working,
452 Err(error) => return fail(error, backlog, requests),478 Err(error) => return fail(&shared, error, backlog, requests),
453 };479 };
454 match crate::task::complete(working.serve(&shared, &requests, &mut backlog)) {480 match crate::task::complete(working.serve(&shared, &requests, &mut backlog)) {
455 Next::Stop => {}481 Next::Stop => {}
...@@ -479,6 +505,34 @@ fn inline(shared: &Thread, job: Job) -> bool {...@@ -479,6 +505,34 @@ fn inline(shared: &Thread, job: Job) -> bool {
479 true505 true
480}506}
481507
508/// Writes again each second the edits a dropped replica left waiting, until none do or their
509/// cache is gone, keeping the cache open and `Released` from firing meanwhile.
510#[cfg(not(target_arch = "wasm32"))]
511fn orphan(shared: &Arc<Thread>, unwritten: Vec<Accepted>) {
512 let shared = Arc::clone(shared);
513 let path = lock(&shared.connection)
514 .ok()
515 .and_then(|connection| connection.path().map(PathBuf::from));
516 let spawned = std::thread::Builder::new()
517 .name("onestore-unwritten".into())
518 .spawn(move || {
519 let arena = Arena::default();
520 let Ok(mut working) = Working::open(&arena, &shared.connection, shared.key.as_ref())
521 else {
522 return;
523 };
524 working.unwritten = unwritten;
525 // A cache deleted, as with its notebook, takes no edits.
526 while !working.unwritten.is_empty() && path.as_ref().is_some_and(|path| path.exists()) {
527 std::thread::sleep(std::time::Duration::from_secs(1));
528 let _ = working.flush(&shared);
529 }
530 });
531 if let Err(error) = spawned {
532 eprintln!("Edits waiting for the cache are lost: {error}");
533 }
534}
535
482/// An edit to apply, with who waits for it.536/// An edit to apply, with who waits for it.
483pub(crate) struct Accepted {537pub(crate) struct Accepted {
484 pub author: String,538 pub author: String,
...@@ -532,6 +586,10 @@ impl<'a> Working<'a> {...@@ -532,6 +586,10 @@ impl<'a> Working<'a> {
532 Err(mpsc::RecvTimeoutError::Timeout) => write = true,586 Err(mpsc::RecvTimeoutError::Timeout) => write = true,
533 Err(mpsc::RecvTimeoutError::Disconnected) => {587 Err(mpsc::RecvTimeoutError::Disconnected) => {
534 let _ = self.flush(shared);588 let _ = self.flush(shared);
589 #[cfg(not(target_arch = "wasm32"))]
590 if !self.unwritten.is_empty() {
591 orphan(shared, std::mem::take(&mut self.unwritten));
592 }
535 return Next::Stop;593 return Next::Stop;
536 }594 }
537 }595 }
...@@ -810,6 +868,7 @@ impl<'a> Working<'a> {...@@ -810,6 +868,7 @@ impl<'a> Working<'a> {
810 }) {868 }) {
811 (edit.reply)(Err(io::Error::other(message.clone()).into()));869 (edit.reply)(Err(io::Error::other(message.clone()).into()));
812 }870 }
871 shared.wait(self.unwritten.len());
813 if let Some(notice) = self872 if let Some(notice) = self
814 .unwritten873 .unwritten
815 .first()874 .first()
...@@ -845,6 +904,7 @@ impl<'a> Working<'a> {...@@ -845,6 +904,7 @@ impl<'a> Working<'a> {
845 }904 }
846 crate::edited(&shared.worker);905 crate::edited(&shared.worker);
847 }906 }
907 shared.wait(0);
848 Ok(())908 Ok(())
849 }909 }
850910
crates/snowbound/src/library.rs+76
...@@ -1924,6 +1924,82 @@ mod tests {...@@ -1924,6 +1924,82 @@ mod tests {
1924 notebook::fs::remove_dir_all(&root).unwrap();1924 notebook::fs::remove_dir_all(&root).unwrap();
1925 }1925 }
19261926
1927 /// An edit the cache can't write yet outlives switching away from its section, counts
1928 /// toward quitting, and is in the cache once the cache takes it, as a cold open reads.
1929 #[test]
1930 fn an_edit_waiting_for_the_cache_outlives_its_section() {
1931 use onestore::op::{Edit, Op, SectionOp};
1932
1933 let root = std::env::temp_dir().join(format!("snowbound-waiting-{}", std::process::id()));
1934 let _ = notebook::fs::remove_dir_all(&root);
1935 let folder = root.join("Personal");
1936 notebook::fs::create_dir_all(&folder).unwrap();
1937 let sample =
1938 Path::new(env!("CARGO_MANIFEST_DIR")).join("../../docs/sample-notebook/Personal");
1939 for entry in notebook::fs::read_dir(sample).unwrap() {
1940 let entry = entry.unwrap();
1941 notebook::fs::copy(entry.path(), folder.join(entry.file_name())).unwrap();
1942 }
1943 let library = Library::notebook(folder.to_str().unwrap(), &root.join("cache"));
1944 let paths: Vec<String> = library.tabs("").into_iter().map(|tab| tab.path).collect();
1945 let first = &paths[0];
1946 let notebook = library.notebook.as_ref().unwrap().as_ref().unwrap();
1947 let cache = notebook.replica_path(first).unwrap();
1948 library.open(first, || {}).unwrap().close().unwrap();
1949 // The cache takes no writes from here until the test lets it. Holding the lock keeps
1950 // every other test's cache writing, so the process counts this test's edit alone.
1951 crate::vfs::install();
1952 let _exclusive =
1953 (crate::vfs::EXCLUSIVE.lock()).unwrap_or_else(std::sync::PoisonError::into_inner);
1954 let deadline = Instant::now() + Duration::from_secs(10);
1955 while notebook::unwritten() > 0 {
1956 assert!(Instant::now() < deadline, "another test's edits still wait");
1957 std::thread::sleep(Duration::from_millis(10));
1958 }
1959 *crate::vfs::FAILING.lock().unwrap() = Some(cache.canonicalize().unwrap());
1960 let section = library.open(first, || {}).unwrap();
1961 let creation = onestore::PageCreation::new(None, Some("Waiting"), "Fixture").unwrap();
1962 let ops = vec![Op::Section(SectionOp::Create(creation))];
1963 let at = crate::filetime();
1964 section.apply("Fixture", Edit { at, ops }).unwrap();
1965 let deadline = Instant::now() + Duration::from_secs(10);
1966 while notebook::unwritten() == 0 {
1967 assert!(Instant::now() < deadline, "the cache took the edit");
1968 std::thread::sleep(Duration::from_millis(10));
1969 }
1970
1971 // Switching through every other section closes the first and drops its replica.
1972 let replica = Arc::downgrade(section.replica());
1973 library.keep(first, section);
1974 for path in &paths[1..] {
1975 library.keep(path, library.open(path, || {}).unwrap());
1976 }
1977 while replica.strong_count() > 0 {
1978 assert!(Instant::now() < deadline, "the section stayed open");
1979 std::thread::sleep(Duration::from_millis(10));
1980 }
1981 assert_eq!(notebook::unwritten(), 1);
1982
1983 *crate::vfs::FAILING.lock().unwrap() = None;
1984 while notebook::unwritten() > 0 {
1985 assert!(Instant::now() < deadline, "the edit was not written");
1986 std::thread::sleep(Duration::from_millis(10));
1987 }
1988 let cold = loop {
1989 match notebook::Replica::open(&cache) {
1990 Ok(cold) => break cold,
1991 Err(error) if error.busy() && Instant::now() < deadline => {
1992 std::thread::sleep(Duration::from_millis(10));
1993 }
1994 Err(error) => panic!("{error}"),
1995 }
1996 };
1997 let pages = cold.pages().unwrap();
1998 assert!(pages.iter().any(|(_, title, _)| title == "Waiting"));
1999 drop((cold, library));
2000 notebook::fs::remove_dir_all(&root).unwrap();
2001 }
2002
1927 #[cfg(unix)]2003 #[cfg(unix)]
1928 #[test]2004 #[test]
1929 fn within_mount_resolves_symlinks() {2005 fn within_mount_resolves_symlinks() {
crates/snowbound/src/main.rs+42-24
...@@ -6544,35 +6544,28 @@ impl State {...@@ -6544,35 +6544,28 @@ impl State {
6544}6544}
65456545
6546#[cfg(not(target_arch = "wasm32"))]6546#[cfg(not(target_arch = "wasm32"))]
6547impl App {6547impl State {
6548 fn close(&self, event_loop: &ActiveEventLoop) {6548 /// Quits, first asking whether to discard a temporary page holding edits.
6549 let Some(state) = self.state.as_ref().filter(|state| {6549 fn quit(&self) {
6550 let editor = &state.view.editor;6550 let editor = &self.view.editor;
6551 let unchanged = editor.caret_outline().is_none_or(TextOutline::is_empty)6551 let unchanged = editor.caret_outline().is_none_or(TextOutline::is_empty)
6552 && state.initial_date == editor.date().map(|date| date.timestamp())6552 && self.initial_date == editor.date().map(|date| date.timestamp())
6553 && state6553 && (self.initial_layouts.iter())
6554 .initial_layouts6554 .map(|(id, layout)| (*id, layout))
6555 .iter()6555 .eq(editor.object_layouts())
6556 .map(|(id, layout)| (*id, layout))6556 && (self.initial.iter())
6557 .eq(editor.object_layouts())6557 .map(|(id, document)| (id, document))
6558 && state6558 .eq((editor.outlines().iter()).map(|outline| (&outline.id, outline.document())));
6559 .initial6559 if self.session.is_some() || unchanged {
6560 .iter()6560 let _ = self.proxy.send_event(UserEvent::Exit);
6561 .map(|(id, document)| (id, document))6561 return;
6562 .eq(editor6562 }
6563 .outlines()
6564 .iter()
6565 .map(|outline| (&outline.id, outline.document())));
6566 state.session.is_none() && !unchanged
6567 }) else {
6568 return event_loop.exit();
6569 };
6570 crate::confirm(6563 crate::confirm(
6571 "Discard this page?",6564 "Discard this page?",
6572 "This temporary page has no saved copy. Closing it will discard your edits.",6565 "This temporary page has no saved copy. Closing it will discard your edits.",
6573 "Keep Editing",6566 "Keep Editing",
6574 "Discard Changes",6567 "Discard Changes",
6575 state.reply(|state, ()| {6568 self.reply(|state, ()| {
6576 let _ = state.proxy.send_event(UserEvent::Exit);6569 let _ = state.proxy.send_event(UserEvent::Exit);
6577 Ok(())6570 Ok(())
6578 }),6571 }),
...@@ -6580,6 +6573,31 @@ impl App {...@@ -6580,6 +6573,31 @@ impl App {
6580 }6573 }
6581}6574}
65826575
6576#[cfg(not(target_arch = "wasm32"))]
6577impl App {
6578 fn close(&self, event_loop: &ActiveEventLoop) {
6579 let Some(state) = &self.state else {
6580 return event_loop.exit();
6581 };
6582 let changes = match notebook::unwritten() {
6583 0 => return state.quit(),
6584 1 => "1 unsaved change".to_owned(),
6585 count => format!("{count} unsaved changes"),
6586 };
6587 crate::confirm(
6588 &format!("Quit with {changes}?"),
6589 "Snowbound keeps trying to save while it’s open. Quitting now loses what hasn’t \
6590 been saved.",
6591 "Keep Snowbound Open",
6592 "Quit Anyway",
6593 state.reply(|state, ()| {
6594 state.quit();
6595 Ok(())
6596 }),
6597 );
6598 }
6599}
6600
6583/// How long quitting waits for the open section's edits to reach its file and for every6601/// How long quitting waits for the open section's edits to reach its file and for every
6584/// section to close; edits left stay in the replica and publish at the next launch.6602/// section to close; edits left stay in the replica and publish at the next launch.
6585#[cfg(not(target_arch = "wasm32"))]6603#[cfg(not(target_arch = "wasm32"))]