diff --git a/apps/ios/Snowbound/Notebooks.swift b/apps/ios/Snowbound/Notebooks.swift index fa63d3af34b7e6ba431727d9b010340d4dff81a1..c52ccb8a9ee9744ddc79b18a783b38956ed27e3f 100644 --- a/apps/ios/Snowbound/Notebooks.swift +++ b/apps/ios/Snowbound/Notebooks.swift @@ -636,6 +636,8 @@ final class Section { static let rejected: UInt32 = 4 fileprivate static var open: [Int: Weak
] = [:] + /// Sections reporting Not Saving, kept open so that edits waiting in memory stay with them. + private static var unsaved: [Int: Section] = [:] private static var tokens = 0 let notebook: Notebook @@ -738,6 +740,7 @@ final class Section { let unlisted = flags & ~Self.listed let unchanged = unlisted == 0 && changed == self.status self.status = changed + Self.unsaved[token] = changed == .notSaving ? self : nil NotificationCenter.default.post( name: Self.changed, object: self, userInfo: unchanged ? ["flags": unlisted, "unchanged": true] : ["flags": unlisted]) diff --git a/crates/mobile/src/library.rs b/crates/mobile/src/library.rs index b8b1a6ce169bfeceeffc07e81ba90d93ce9d4e63..40a615c8fdad0f71dd154efb6fdb8fae606ca456 100644 --- a/crates/mobile/src/library.rs +++ b/crates/mobile/src/library.rs @@ -1098,7 +1098,7 @@ impl Section { Status::Offline } Event::Failed(error) => { - eprintln!("snowbound: synchronization stopped: {error}"); + eprintln!("snowbound: not saving: {error}"); Status::NotSaving } }; @@ -1284,6 +1284,7 @@ impl Library { let deadline = Instant::now() + limit; let open = self.open_sections(); for (_, shared) in &open { + shared.section.written().map_err(|_| WAITING)?; while !shared.section.pending()?.is_empty() { if Instant::now() >= deadline { return Err(WAITING.into()); diff --git a/crates/mobile/src/tests.rs b/crates/mobile/src/tests.rs index 98db26cbe323b927d58576b880bbf7c38a10ba80..01a62b4d8c196c90a5922ec421af8a7e78498017 100644 --- a/crates/mobile/src/tests.rs +++ b/crates/mobile/src/tests.rs @@ -3,6 +3,9 @@ use library::{sb_library_set_offline, sb_library_sync_now}; use notebook::session::SyncState; use std::path::{Path, PathBuf}; +#[path = "../../notebook/tests/support/vfs.rs"] +mod vfs; + const CORPUS: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/../../corpus"); fn corpus(path: &str) -> PathBuf { @@ -3165,14 +3168,9 @@ fn reader_deferred_reload_preserves_rich_composition_after_real_local_write_fail .section .close() .unwrap(); - let sql = rusqlite::Connection::open(&database).unwrap(); - sql.busy_timeout(Duration::from_secs(1)).unwrap(); - sql.execute_batch( - "CREATE TRIGGER reader_fault BEFORE INSERT ON edits WHEN NEW.author = 'Fixture' - BEGIN SELECT RAISE(ABORT, 'Disposable Reader write failure'); END;", - ) - .unwrap(); - sql.close().unwrap(); + // The cache's WAL takes no writes until the test has seen the edits wait. + vfs::install(); + *vfs::FAILING.lock().unwrap() = Some(database.canonicalize().unwrap()); let library = Arc::new(Library::open(&file, &cache, true).unwrap()); sb_library_set_offline(&library, true); @@ -3183,7 +3181,6 @@ fn reader_deferred_reload_preserves_rich_composition_after_real_local_write_fail || {}, ) .unwrap(); - assert_eq!(section.shared.section.identity().unwrap(), identity); assert_eq!(section.shared.section.pending().unwrap(), seed); let (page, read_only) = section.shared.page(space).unwrap(); let mut view = View { @@ -3204,30 +3201,13 @@ fn reader_deferred_reload_preserves_rich_composition_after_real_local_write_fail ); assert_ne!(sb_view_format(&view) & 1, 0); assert!(unsafe { sb_set_marked(&mut view, c"FAILED rich 😀 ".as_ptr(), 15, 15) }); - let nodes = view.canvas.active().document().nodes().to_vec(); - let preview = view.canvas.page.editor.page().unwrap(); - section - .shared - .section - .apply( - "Reader peer", - page_edit(space, vec![PageOp::Color(Some(0x0011_2233))]), - ) - .unwrap(); - section.shared.section.written().unwrap(); - let before = section.shared.section.pending().unwrap(); - assert_eq!(before.len(), seed.len() + 1); - assert_eq!(&before[..seed.len()], seed.as_slice()); - assert!(!sb_view_reload(&mut view, false, Some(&mut anchor))); - assert!(view.reload_pending); - assert!(!sb_view_reload_ready(&view)); if select { assert!(sb_select(&mut view, 0, 0)); } else { sb_unmark(&mut view); } - let history = view.canvas.page.editor.history_depth(); - let selection = view.canvas.page.editor.selection(); + assert!(sb_view_set_paper(&mut view, 0x77, 0x88, 0x99, -1)); + let preview = view.canvas.page.editor.page().unwrap(); let deadline = Instant::now() + Duration::from_secs(10); let mut flags = 0; while section.shared.status.load(Ordering::Relaxed) != library::Status::NotSaving as u8 { @@ -3237,107 +3217,44 @@ fn reader_deferred_reload_preserves_rich_composition_after_real_local_write_fail } flags |= section.poll(); assert_eq!(flags & 4, 0); - assert!(view.local_edits.lock().unwrap().failed); - assert_eq!(view.canvas.active().document().nodes(), nodes); - assert_eq!(view.canvas.page.editor.page().unwrap(), preview); - assert!(view.canvas.active().shown_text().contains("FAILED rich 😀")); - assert!(view.canvas.edit().unwrap().is_none()); - section.shared.section.written().unwrap(); - assert_eq!(section.shared.section.pending().unwrap(), before); - assert!(!sb_view_reload(&mut view, false, Some(&mut anchor))); - assert!(!sb_view_reload_ready(&view)); - assert!(view.reload_pending); - assert_eq!(view.canvas.page.editor.page().unwrap(), preview); - assert_eq!(view.canvas.page.editor.history_depth(), history); - assert_eq!(view.canvas.page.editor.selection(), selection); - assert_eq!(std::fs::read(&file).unwrap(), original); + assert_eq!(section.shared.section.pending().unwrap(), seed); + let (waiting, _) = section.shared.page(space).unwrap(); + assert!(canvas::search::page_text(&waiting).contains("FAILED rich 😀")); + assert_eq!(waiting.color, Some(0x0099_8877)); + assert_eq!(view.local_edits.lock().unwrap().pending, 2); + assert!(!view.local_edits.lock().unwrap().failed); + assert!(library.close(Duration::from_millis(100)).is_err()); + *vfs::FAILING.lock().unwrap() = None; - section - .shared - .section - .apply( - "Reader peer", - page_edit(space, vec![PageOp::Color(Some(0x0044_5566))]), - ) - .unwrap(); - section.shared.section.written().unwrap(); - let _ = section.poll(); - assert!(!sb_view_reload_ready(&view)); - assert!(!sb_view_reload(&mut view, false, None)); + let deadline = Instant::now() + Duration::from_secs(10); + while view.local_edits.lock().unwrap().pending > 0 { + assert!(Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(10)); + } + assert!(!view.local_edits.lock().unwrap().failed); + let pending = section.shared.section.pending().unwrap(); + assert_eq!(pending.len(), seed.len() + 2); + assert_eq!(&pending[..seed.len()], seed.as_slice()); + sb_view_reload(&mut view, false, None); + assert!(!view.reload_pending); assert_eq!(view.canvas.page.editor.page().unwrap(), preview); - unsafe { library::sb_section_set_author(§ion, c"Reader peer".as_ptr()) }; - assert!(sb_view_set_paper(&mut view, 0x77, 0x88, 0x99, -1)); - section.shared.section.written().unwrap(); - let successful_later = section.shared.section.pending().unwrap(); - assert_eq!(successful_later.len(), before.len() + 2); - assert_eq!(successful_later.last().unwrap().author, "Reader peer"); - assert!(view.local_edits.lock().unwrap().failed); - assert!(!sb_view_reload_ready(&view)); - assert!(view.canvas.active().shown_text().contains("FAILED rich 😀")); - let outstanding = ApplyReply::new(&view.local_edits); - let (returned, returned_reply) = std::sync::mpsc::channel(); - section - .shared - .section - .apply_with( - "Fixture", - page_edit(space, vec![PageOp::Color(Some(0x0010_2030))]), - move |result| returned.send((outstanding, result)).unwrap(), - ) - .unwrap(); - let (outstanding, result) = returned_reply - .recv_timeout(Duration::from_secs(10)) - .unwrap(); - assert!(result.is_err()); - assert_eq!(section.shared.section.pending().unwrap(), successful_later); - sb_view_reload(&mut view, true, None); - assert!(!view.canvas.active().shown_text().contains("FAILED rich 😀")); - assert_eq!(view.local_edits.lock().unwrap().pending, 1); - assert!(!view.local_edits.lock().unwrap().failed); - sb_view_reload(&mut view, false, None); - assert!(!sb_view_reload_ready(&view)); - outstanding.complete(result); - assert!(view.local_edits.lock().unwrap().failed); - assert!(!sb_view_reload_ready(&view)); - sb_view_reload(&mut view, true, None); - assert!(!view.local_edits.lock().unwrap().failed); - assert_eq!(view.local_edits.lock().unwrap().pending, 0); - assert!(!view.reload_pending); - let _ = section.poll(); - assert!(sb_view_set_paper(&mut view, 0x11, 0x22, 0x33, -1)); - section.shared.section.written().unwrap(); - section - .shared - .section - .apply( - "Reader peer", - page_edit(space, vec![PageOp::Color(Some(0x0012_3456))]), - ) - .unwrap(); - section.shared.section.written().unwrap(); - sb_view_reload(&mut view, false, None); - assert!(!view.reload_pending); - assert!(!view.local_edits.lock().unwrap().failed); - assert_eq!( - view.canvas.page.editor.page().unwrap(), - section.shared.page(space).unwrap().0 - ); - assert_eq!(std::fs::read(&file).unwrap(), original); - let preserved = section.shared.section.pending().unwrap(); let shared = Arc::clone(§ion.shared); drop((view, section, library)); Arc::into_inner(shared) - .expect("No other failed writer") + .expect("No other writer") .section .close() .unwrap(); - let sql = rusqlite::Connection::open(&database).unwrap(); - sql.busy_timeout(Duration::from_secs(1)).unwrap(); - sql.execute_batch("DROP TRIGGER reader_fault").unwrap(); - sql.close().unwrap(); let cold = notebook::Replica::open(&database).unwrap(); assert_eq!(cold.identity().unwrap(), identity); - assert_eq!(cold.pending().unwrap(), preserved); - assert!(!canvas::search::page_text(&cold.page(space).unwrap()).contains("FAILED rich 😀")); + assert_eq!(cold.pending().unwrap(), pending); + let page = cold.page(space).unwrap(); + assert_eq!( + canvas::search::page_text(&page) + .matches("FAILED rich 😀") + .count(), + 1 + ); + assert_eq!(page.color, Some(0x0099_8877)); } } diff --git a/crates/notebook/README.md b/crates/notebook/README.md index c07caced6ae3bd69dfd18f64f0b310aaffa03dcb..2ce910c1fa02384de9a5546e4a183dfe6c75eabf 100644 --- a/crates/notebook/README.md +++ b/crates/notebook/README.md @@ -44,8 +44,11 @@ it was, then hands its requests over; edits sent meanwhile apply to the rebuilt 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. After a database error the section -thread rereads the queue. +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. ## Sessions diff --git a/crates/notebook/src/lib.rs b/crates/notebook/src/lib.rs index 3fdc43864fe577bb6bf791c51731c4031d6263ab..3b3c96854967a9d6053de89da4ce4897ca175d8e 100644 --- a/crates/notebook/src/lib.rs +++ b/crates/notebook/src/lib.rs @@ -302,26 +302,32 @@ impl Replica { /// Applies an edit, awaiting its durable local queue receipt without blocking browser reads. pub async fn apply_async(&self, author: &str, edit: Edit) -> Result { - self.ask_async(|reply| working::Request::Apply { - author: author.to_owned(), - edit, - reply, + self.ask_async(|reply| { + working::Request::Apply(working::Accepted { + author: author.to_owned(), + edit, + reply, + waiting: None, + }) }) .await } - /// `apply` without waiting: `reply` runs once the edit is durable or refused. + /// `apply` without waiting: `reply` runs once the edit is durable or refused. While the + /// cache refuses to write it, the edit waits in memory and each failed write tells `waiting`. pub(crate) fn submit( &self, author: &str, edit: Edit, reply: working::Reply, + waiting: working::Notice, ) -> Result<()> { - self.send(working::Request::Apply { + self.send(working::Request::Apply(working::Accepted { author: author.to_owned(), edit, reply, - }) + waiting: Some(waiting), + })) } /// The page in `space` as the queued edits leave it; O(page), for opening and reloading. diff --git a/crates/notebook/src/session.rs b/crates/notebook/src/session.rs index e08d7a8470b4b6fb82cd903c74418d5958846172..9ad4304e16ba3e6ee362794e02682fa66d0d5714 100644 --- a/crates/notebook/src/session.rs +++ b/crates/notebook/src/session.rs @@ -2136,7 +2136,7 @@ pub enum Event { Attempt { id: u64, status: EditStatus }, /// The section file could not be reached; the replica keeps its state. Unreachable(io::Error), - /// The replica itself failed; the worker has stopped. + /// The replica itself failed and the worker stopped, or edits wait in memory for the cache. Failed(String), } @@ -2482,6 +2482,14 @@ impl Section { reply: impl FnOnce(Result) + Send + 'static, ) -> Result<()> { let (sender, notify) = (self.sender.clone(), Arc::clone(&self.notify)); + let waiting: crate::working::Notice = { + let (sender, notify) = (sender.clone(), Arc::clone(¬ify)); + Box::new(move |error| { + if sender.send(Event::Failed(error.to_owned())).is_ok() { + notify(); + } + }) + }; let root = self.replica.root; let spaces = crate::queue::spaces(&edit, root); self.replica.submit( @@ -2505,6 +2513,7 @@ impl Section { notify(); } }), + waiting, ) } diff --git a/crates/notebook/src/session/tests.rs b/crates/notebook/src/session/tests.rs index 69ac021f7e0e5bd774cbb1400da20b80860b7fc7..5e648b2a75cc8bb2e058e9cf41f85714566620c4 100644 --- a/crates/notebook/src/session/tests.rs +++ b/crates/notebook/src/session/tests.rs @@ -670,3 +670,186 @@ fn associated_apply_reply_precedes_success_wake_without_a_publication_event() { section.close().unwrap(); assert_eq!(fs::read(file).unwrap(), source); } + +#[test] +fn edits_the_cache_refuses_wait_in_order_and_land_once() { + use std::time::Instant; + + let temporary = tempfile::tempdir().unwrap(); + let file = temporary.path().join("Waiting.one"); + let cache = temporary.path().join("waiting.sqlite"); + let source = onestore::create_section("Waiting.one", "Original", "Fixture").unwrap(); + fs::write(&file, &source).unwrap(); + let replica = Replica::create(&cache, &source).unwrap(); + let section = Section::resume(&file, replica, true, || {}).unwrap(); + let replica = Arc::clone(section.replica()); + // Queued before the fault, so the worker asks to seal while edits wait. + let seed = replica + .apply( + "Fixture", + Edit { + at: crate::now(), + ops: vec![Op::Section(SectionOp::Create( + PageCreation::new(None, Some("Seed"), "Fixture").unwrap(), + ))], + }, + ) + .unwrap(); + // The cache is full, until it may grow again; the first edit needs room. + let full = |pages: Option| { + let connection = replica.lock().unwrap(); + let pages = pages.unwrap_or_else(|| { + connection + .pragma_query_value(None, "page_count", |row| row.get(0)) + .unwrap() + }); + connection + .pragma_update(None, "max_page_count", pages) + .unwrap(); + }; + full(None); + let title = "Waiting page ".repeat(1000); + let creation = PageCreation::new(None, Some(&title), "Fixture").unwrap(); + let space = creation.space(); + // The second edit colours the page the first creates. + let edits = [ + Op::Section(SectionOp::Create(creation)), + Op::Page { + space, + op: onestore::op::PageOp::Color(Some(0x00ab_cdef)), + }, + ] + .map(|op| Edit { + at: crate::now(), + ops: vec![op], + }); + let (replied, replies) = mpsc::channel(); + for edit in edits.clone() { + let replied = replied.clone(); + section + .apply_with("Fixture", edit, move |result| { + replied + .send(result.map_err(|error| error.to_string())) + .unwrap(); + }) + .unwrap(); + } + let deadline = Instant::now() + Duration::from_secs(10); + while !section + .events() + .iter() + .any(|event| matches!(event, Event::Failed(_))) + { + assert!(Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(10)); + } + let waiting = |section: &Section| { + assert!(replies.try_recv().is_err()); + let pending = section.pending().unwrap(); + assert_eq!( + pending.iter().map(|edit| edit.id).collect::>(), + [seed] + ); + let page = section.page(space).unwrap(); + assert_eq!((&page.title, page.color), (&title, Some(0x00ab_cdef))); + assert!(matches!( + replica.ask(|reply| crate::working::Request::Seal { reply }), + Err(Error::Io(error)) if error.kind() == io::ErrorKind::WouldBlock + )); + assert!(matches!( + section.written(), + Err(Error::Io(error)) if error.kind() == io::ErrorKind::WouldBlock + )); + }; + waiting(§ion); + // A caller blocking on its edit still learns of the failure, and the waiting edits stay. + let blocked = Edit { + at: crate::now(), + ops: vec![Op::Page { + space, + op: onestore::op::PageOp::Color(Some(0x0012_3456)), + }], + }; + assert!(replica.apply("Fixture", blocked).is_err()); + waiting(§ion); + section.wake(); + std::thread::sleep(Duration::from_millis(200)); + waiting(§ion); + + full(Some(1 << 30)); + let ids = [(); 2].map(|()| { + replies + .recv_timeout(Duration::from_secs(10)) + .unwrap() + .unwrap() + }); + let pending = section.pending().unwrap(); + assert_eq!( + pending[1..] + .iter() + .map(|edit| (edit.id, &edit.edit)) + .collect::>(), + [(ids[0], &edits[0]), (ids[1], &edits[1])] + ); + + // An edit that could never be written is refused alone; the next one lands. + replica + .lock() + .unwrap() + .execute_batch( + "CREATE TEMP TRIGGER refuse BEFORE INSERT ON main.edits WHEN NEW.author = 'Refused' + BEGIN SELECT RAISE(ABORT, 'Disposable refusal'); END;", + ) + .unwrap(); + for (author, colour) in [("Refused", 0x0012_3456), ("Fixture", 0x0065_4321)] { + let replied = replied.clone(); + section + .apply_with( + author, + Edit { + at: crate::now(), + ops: vec![Op::Page { + space, + op: onestore::op::PageOp::Color(Some(colour)), + }], + }, + move |result| { + replied + .send(result.map_err(|error| error.to_string())) + .unwrap(); + }, + ) + .unwrap(); + } + assert!( + replies + .recv_timeout(Duration::from_secs(10)) + .unwrap() + .is_err() + ); + let last = replies + .recv_timeout(Duration::from_secs(10)) + .unwrap() + .unwrap(); + section.written().unwrap(); + assert_eq!(section.pending().unwrap().last().unwrap().id, last); + assert_eq!(section.page(space).unwrap().color, Some(0x0065_4321)); + section.wake(); + let deadline = Instant::now() + Duration::from_secs(10); + while !section.events().iter().any(|event| { + matches!(event, Event::Attempt { id, status: EditStatus::Published { .. } } if *id == last) + }) { + assert!(Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(10)); + } + section.close().unwrap(); + drop(replica); + let pages = stored_pages(&fs::read(&file).unwrap()).unwrap(); + let waited: Vec<_> = pages + .iter() + .filter(|page| page.page.title == title) + .collect(); + assert_eq!(waited.len(), 1); + assert_eq!(waited[0].page.color, Some(0x0065_4321)); + assert!(Replica::open(&cache).unwrap().pending().unwrap().is_empty()); +} diff --git a/crates/notebook/src/working.rs b/crates/notebook/src/working.rs index 0c76eabb12fdd3825cf7b331e95abde8b091cab3..b9a0a33d50eabd6b1859f0cec327585ba7931b9b 100644 --- a/crates/notebook/src/working.rs +++ b/crates/notebook/src/working.rs @@ -15,17 +15,15 @@ use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params}; use std::{ collections::{BTreeMap, BTreeSet, VecDeque}, io, + path::PathBuf, sync::{Arc, Mutex, Weak, mpsc}, }; pub(crate) type Reply = Box) + Send>; +pub(crate) type Notice = Box; pub(crate) enum Request { - Apply { - author: String, - edit: Edit, - reply: Reply, - }, + Apply(Accepted), Page { space: ExGuid, reply: Reply, @@ -82,10 +80,10 @@ pub(crate) struct Sealed { } impl Request { - fn fail(self, message: &str) { - let error = || io::Error::other(message.to_owned()).into(); + fn fail(self, kind: io::ErrorKind, message: &str) { + let error = || io::Error::new(kind, message.to_owned()).into(); match self { - Self::Apply { reply, .. } => reply(Err(error())), + Self::Apply(Accepted { reply, .. }) => reply(Err(error())), Self::Page { reply, .. } => reply(Err(error())), Self::Pages { reply } => reply(Err(error())), Self::Conflicts { reply } => reply(Err(error())), @@ -288,19 +286,33 @@ mod serving { } } -/// The next request, or none once the replica dropped. +/// The next request; a second passes without one while `retry`. #[cfg(not(target_arch = "wasm32"))] -async fn receive(requests: &mpsc::Receiver) -> Option { - requests.recv().ok() +async fn receive( + requests: &mpsc::Receiver, + retry: bool, +) -> std::result::Result { + if retry { + requests.recv_timeout(std::time::Duration::from_secs(1)) + } else { + requests + .recv() + .map_err(|_| mpsc::RecvTimeoutError::Disconnected) + } } -/// The next request, waiting for `serving::serve` where none is queued, or none once the -/// replica dropped. +/// The next request, waiting for `serving::serve` where none is queued; requests alone +/// retry unwritten edits. #[cfg(target_arch = "wasm32")] -async fn receive(requests: &mpsc::Receiver) -> Option { +async fn receive( + requests: &mpsc::Receiver, + _retry: bool, +) -> std::result::Result { std::future::poll_fn(|_| match requests.try_recv() { - Ok(request) => std::task::Poll::Ready(Some(request)), - Err(mpsc::TryRecvError::Disconnected) => std::task::Poll::Ready(None), + Ok(request) => std::task::Poll::Ready(Ok(request)), + Err(mpsc::TryRecvError::Disconnected) => { + std::task::Poll::Ready(Err(mpsc::RecvTimeoutError::Disconnected)) + } Err(mpsc::TryRecvError::Empty) => std::task::Poll::Pending, }) .await @@ -386,7 +398,7 @@ async fn run( fn fail(error: crate::Error, backlog: VecDeque, requests: mpsc::Receiver) { let message = error.to_string(); for request in backlog.into_iter().chain(requests.iter()) { - request.fail(&message); + request.fail(io::ErrorKind::Other, &message); } } @@ -462,11 +474,13 @@ fn inline(shared: &Thread, job: Job) -> bool { true } -/// An edit applied to the section and not yet written, with who waits for it. -struct Accepted { - author: String, - edit: Edit, - reply: Reply, +/// An edit to apply, with who waits for it. +pub(crate) struct Accepted { + pub author: String, + pub edit: Edit, + pub reply: Reply, + /// Set where nobody blocks on the reply: a failed write keeps the edit and reports here. + pub waiting: Option, } struct Working<'a> { @@ -475,6 +489,9 @@ struct Working<'a> { open: Option, /// Spaces the open batch's edits change. touched: BTreeSet, + /// Applied to the section and not yet in the queue, so the section is the queue replayed + /// with these after it; written in order before anything seals, rebases or rereads. + unwritten: Vec, } impl<'a> Working<'a> { @@ -484,6 +501,7 @@ impl<'a> Working<'a> { section, open, touched, + unwritten: Vec::new(), }) } @@ -498,20 +516,24 @@ impl<'a> Working<'a> { // other request waits for the rebuilt section. let mut held: Option> = None; loop { - let first = match backlog.pop_front() { - Some(request) => request, - None => match receive(requests).await { - Some(request) => request, - None => return Next::Stop, - }, - }; - let mut burst: VecDeque = VecDeque::from([first]); - burst.extend(backlog.drain(..)); + let mut burst = std::mem::take(backlog); + // Only a burst that took an edit, or a retry's tick, writes at its end: reads never. + let mut write = false; + if burst.is_empty() { + match receive(requests, !self.unwritten.is_empty()).await { + Ok(request) => burst.push_back(request), + Err(mpsc::RecvTimeoutError::Timeout) => write = true, + Err(mpsc::RecvTimeoutError::Disconnected) => { + let _ = self.flush(shared); + return Next::Stop; + } + } + } burst.extend(requests.try_iter()); - let mut accepted = Vec::new(); // Requests that write the queue beyond the burst's edits run after its reads. let mut later = VecDeque::new(); - while let Some(request) = burst.pop_front().or_else(|| later.pop_front()) { + let mut reopen = false; + while !reopen && let Some(request) = burst.pop_front().or_else(|| later.pop_front()) { if let Some(waiting) = &mut held { match request { Request::Page { space, reply } => { @@ -542,15 +564,29 @@ impl<'a> Working<'a> { } continue; } - let reopen = match request { - Request::Apply { - author, - edit, - reply, - } => self - .accept(author, edit, reply, &mut accepted, shared) - .err() - .unwrap_or(false), + if matches!(request, Request::Flush { .. } | Request::Seal { .. }) + && !burst.is_empty() + { + later.push_back(request); + continue; + } + if matches!( + request, + Request::Flush { .. } | Request::Seal { .. } | Request::Rebase { .. } + ) { + let failed = self.flush(shared).is_err(); + if failed || !self.unwritten.is_empty() { + // The worker retries later; the edits wait here meanwhile. + request.fail(io::ErrorKind::WouldBlock, "Edits wait to be written"); + reopen = failed; + continue; + } + } + reopen = match request { + Request::Apply(accepted) => { + write = true; + self.accept(accepted) + } Request::Page { space, reply } => { reply(self.section.page(space).map_err(Into::into)); false @@ -575,86 +611,62 @@ impl<'a> Working<'a> { reply(self.section.version(space, version).map_err(Into::into)); false } - request @ (Request::Flush { .. } | Request::Seal { .. }) - if !burst.is_empty() => - { - later.push_back(request); - false - } Request::Flush { reply, durable } => { - let written = self.flush(shared, &mut accepted); - if written { - #[cfg(target_arch = "wasm32")] - if durable { - shared.receipt(vec![reply], false); - } else { - reply(Ok(())); - } - #[cfg(not(target_arch = "wasm32"))] - { - let _ = durable; - reply(Ok(())); - } + #[cfg(target_arch = "wasm32")] + if durable { + shared.receipt(vec![reply], false); } else { - reply(Err( - io::Error::other("The queue could not be written").into() - )); + reply(Ok(())); } - !written + #[cfg(not(target_arch = "wasm32"))] + { + let _ = durable; + reply(Ok(())); + } + false } Request::Seal { reply } => { - if !self.flush(shared, &mut accepted) { - reply(Err( - io::Error::other("The queue could not be written").into() - )); - true - } else { - let sealed = self.seal(connection); - let failed = sealed.is_err(); - reply(sealed); - failed - } + let sealed = self.seal(connection); + let failed = sealed.is_err(); + reply(sealed); + failed } Request::Rebase { image, reply } => { - if self.flush(shared, &mut accepted) { - #[cfg(target_arch = "wasm32")] - let reopen = inline(shared, Job::Rebase { image, reply }); - #[cfg(not(target_arch = "wasm32"))] - let reopen = { - held = self - .rebuild(shared, Job::Rebase { image, reply }) - .then(VecDeque::new); - false - }; - reopen - } else { - reply(Err( - io::Error::other("The queue could not be written").into() - )); - true - } + #[cfg(target_arch = "wasm32")] + let reopen = inline(shared, Job::Rebase { image, reply }); + #[cfg(not(target_arch = "wasm32"))] + let reopen = { + held = self + .rebuild(shared, Job::Rebase { image, reply }) + .then(VecDeque::new); + false + }; + reopen } Request::Reopen { reply } => { - self.flush(shared, &mut accepted); + let _ = self.flush(shared); #[cfg(target_arch = "wasm32")] let reopen = inline(shared, Job::Reopen { reply }); #[cfg(not(target_arch = "wasm32"))] let reopen = { held = self .rebuild(shared, Job::Reopen { reply }) - .then(VecDeque::new); + .then(|| self.unwritten.drain(..).map(Request::Apply).collect()); false }; reopen } Request::Handover(_) | Request::Resume => false, }; - if reopen { - backlog.extend(burst.drain(..).chain(later.drain(..))); - return Next::Reopen; - } } - if !self.flush(shared, &mut accepted) { + if reopen || write && self.flush(shared).is_err() { + backlog.extend( + self.unwritten + .drain(..) + .map(Request::Apply) + .chain(burst.drain(..)) + .chain(later.drain(..)), + ); return Next::Reopen; } } @@ -686,101 +698,120 @@ impl<'a> Working<'a> { } /// Applies an edit, keeping it to be written with the burst. A refused edit is answered - /// here; `Err(true)` when it failed part way and the section must be reread. - fn accept( - &mut self, - author: String, - edit: Edit, - reply: Reply, - accepted: &mut Vec, - shared: &Arc, - ) -> std::result::Result<(), bool> { - match self.section.apply(&author, &edit) { + /// here; true when it failed part way and the section must be reread. + fn accept(&mut self, accepted: Accepted) -> bool { + match self.section.apply(&accepted.author, &accepted.edit) { Ok(()) => { self.touched - .extend(queue::spaces(&edit, self.section.root())); - accepted.push(Accepted { - author, - edit, - reply, - }); - Ok(()) + .extend(queue::spaces(&accepted.edit, self.section.root())); + self.unwritten.push(accepted); + false } Err(error) => { let broken = matches!(error, OpError::Failed(_)); - if broken { - self.flush(shared, accepted); - } - reply(Err(error.into())); - Err(broken) + (accepted.reply)(Err(error.into())); + broken } } } - /// Writes the accepted edits in one transaction, then answers their senders; false - /// when the write failed and the section holds edits the queue lacks. - fn flush(&mut self, shared: &Arc, accepted: &mut Vec) -> bool { - if accepted.is_empty() { - return true; + /// Writes the unwritten edits in one transaction, then answers their senders. A write + /// that may succeed later keeps the edits with a notice, telling the first; one that never + /// will refuses the edit it failed on. An error means the section holds an edit the queue + /// never will, or the cache closed, and must be reread. + fn flush(&mut self, shared: &Arc) -> Result<()> { + if self.unwritten.is_empty() { + return Ok(()); } - let written = (|| -> Result> { - let mut connection = lock(&shared.connection)?; - let transaction = - connection.transaction_with_behavior(TransactionBehavior::Immediate)?; - let batch = match self.open { - Some(batch) => batch, - None => { - transaction.execute("INSERT INTO batches DEFAULT VALUES", [])?; - transaction.last_insert_rowid() - } - }; - let mut ids = Vec::new(); - for edit in accepted.iter() { - ids.push(queue::insert( - &transaction, - shared.key.as_ref(), - None, - batch, - &edit.author, - &edit.edit, - )?); - } - transaction.commit()?; - self.open = Some(batch); - Ok(ids) - })(); - let ok = written.is_ok(); - match written { - Ok(ids) => { - #[cfg(target_arch = "wasm32")] - shared.receipt( - accepted - .drain(..) - .zip(ids) - .map(|(edit, id)| { - let reply: Reply<()> = - Box::new(move |result| (edit.reply)(result.map(|()| id))); - reply - }) - .collect(), - true, - ); - #[cfg(not(target_arch = "wasm32"))] - { - for (edit, id) in accepted.drain(..).zip(ids) { - (edit.reply)(Ok(id)); + let mut connection = lock(&shared.connection)?; + let mut ids = Vec::new(); + let committed = connection + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(crate::Error::from) + .and_then(|transaction| { + let batch = match self.open { + Some(batch) => batch, + None => { + transaction.execute("INSERT INTO batches DEFAULT VALUES", [])?; + transaction.last_insert_rowid() } - crate::edited(&shared.worker); + }; + for edit in &self.unwritten { + ids.push(queue::insert( + &transaction, + shared.key.as_ref(), + None, + batch, + &edit.author, + &edit.edit, + )?); + } + Ok((batch, transaction.commit().err())) + }); + // The error, and the edit it refuses when it would fail again. + let written = match committed { + Ok((batch, None)) => Ok(batch), + Ok((batch, Some(error))) => { + if landed(&mut connection, ids.last().copied().unwrap_or_default())? { + Ok(batch) + } else { + Err((error.into(), None)) } } Err(error) => { + let refused = (!transient(&error)).then_some(ids.len()); + Err((error, refused)) + } + }; + drop(connection); + let batch = match written { + Ok(batch) => batch, + Err((error, refused)) => { let message = error.to_string(); - for edit in accepted.drain(..) { + let before = self.unwritten.len(); + let mut at = 0; + for edit in self.unwritten.extract_if(.., |edit| { + at += 1; + edit.waiting.is_none() || refused == Some(at - 1) + }) { (edit.reply)(Err(io::Error::other(message.clone()).into())); } + if let Some(notice) = self + .unwritten + .first() + .and_then(|edit| edit.waiting.as_ref()) + { + notice(&message); + } + return if self.unwritten.len() == before { + Ok(()) + } else { + Err(error) + }; } + }; + self.open = Some(batch); + #[cfg(target_arch = "wasm32")] + shared.receipt( + self.unwritten + .drain(..) + .zip(ids) + .map(|(edit, id)| { + let reply: Reply<()> = + Box::new(move |result| (edit.reply)(result.map(|()| id))); + reply + }) + .collect(), + true, + ); + #[cfg(not(target_arch = "wasm32"))] + { + for (edit, id) in self.unwritten.drain(..).zip(ids) { + (edit.reply)(Ok(id)); + } + crate::edited(&shared.worker); } - ok + Ok(()) } /// Seals the open batch unless a sealed batch still waits for publication. @@ -1052,3 +1083,37 @@ pub(crate) fn sealed(connection: &Connection) -> Result> { }) .transpose() } + +/// Whether a write that failed may succeed when tried again, once the disk has room or a lock +/// is let go; any other error would fail the same way forever. +fn transient(error: &crate::Error) -> bool { + use rusqlite::ErrorCode::*; + matches!(error, crate::Error::Database(rusqlite::Error::SqliteFailure(error, _)) + if matches!(error.code, SystemIoFailure | DiskFull | DatabaseBusy | DatabaseLocked | CannotOpen)) +} + +/// Whether the commit whose last edit is `last` reached the cache after it failed, as a cold +/// open reads it, a commit landing whole: the warm connection may not know of frames its WAL +/// holds, and must never write again. When the cache cannot be reopened, an inert connection +/// stays, failing every statement. +fn landed(connection: &mut Connection, last: u64) -> Result { + let path = connection.path().map(PathBuf::from); + let inert = Connection::open_in_memory()?; + inert.execute_batch("PRAGMA query_only=1")?; + std::mem::replace(connection, inert) + .close() + .map_err(|(_, error)| error)?; + // Reopening through the browser's file system is unverified: it stays closed. + let path = path + .filter(|_| cfg!(not(target_arch = "wasm32"))) + .ok_or_else(|| io::Error::other("The cache could not confirm a write"))?; + let cold = crate::cache_connection(&path)?; + crate::write_ahead(&cold)?; + let found: bool = cold.query_row( + "SELECT EXISTS(SELECT 1 FROM edits WHERE id=?1)", + [crate::signed(last)?], + |row| row.get(0), + )?; + *connection = cold; + Ok(found) +} diff --git a/crates/notebook/tests/commit_fault.rs b/crates/notebook/tests/commit_fault.rs new file mode 100644 index 0000000000000000000000000000000000000000..a6fd97be281971bd15b274183332f4a8a1b62b50 --- /dev/null +++ b/crates/notebook/tests/commit_fault.rs @@ -0,0 +1,177 @@ +//! Commits whose WAL write or sync fails, through a shim over the platform's VFS. Its own +//! binary, as the shim becomes the process's default VFS. +#![cfg(unix)] + +use notebook::{ + Replica, + session::{Event, Section}, +}; +use onestore::op::{Edit, Op, PageOp}; +use std::{path::Path, sync::atomic::Ordering, sync::mpsc, time::Duration}; + +#[path = "support/vfs.rs"] +mod vfs; +use vfs::{ + ARMED, KEEP_WAL, OPEN, SYNC_AFTER, SYNC_BEFORE, WAL_WRITES_AFTER_FAULT, WRITE_COMMIT, install, +}; + +fn colour(space: onestore::ExGuid, colour: u32) -> Edit { + Edit { + at: 133_000_000_000_000_000 + u64::from(colour), + ops: vec![Op::Page { + space, + op: PageOp::Color(Some(colour)), + }], + } +} + +/// What the section answered for the edit committed under `fault` and the one after it, +/// whether a failed write was reported, and the queue as the section saw it. +struct Outcome { + replies: [Result; 2], + failed: bool, + pending: Option>, +} + +/// Seeds the WAL with one edit, so the armed commit's sync is the only one it makes, then +/// commits under `fault` and once more after it; with `crash`, leaves the cache open. +fn scenario(root: &Path, fault: u8, crash: bool) -> Outcome { + install(); + let replica = Replica::open(root.join("replica.sqlite")).unwrap(); + let section = Section::resume(root.join("Fault.one"), replica, true, || {}).unwrap(); + let space = section.pages().unwrap()[0].0; + let apply = |value| { + let (sender, receiver) = mpsc::channel(); + section + .apply_with("Fixture", colour(space, value), move |result| { + sender.send(result).unwrap(); + }) + .unwrap(); + receiver + .recv_timeout(Duration::from_secs(10)) + .unwrap() + .map_err(|error| error.to_string()) + }; + apply(1).unwrap(); + WAL_WRITES_AFTER_FAULT.store(0, Ordering::SeqCst); + ARMED.store(fault, Ordering::SeqCst); + let first = apply(2); + assert_eq!(ARMED.load(Ordering::SeqCst), 0, "fault {fault} never fired"); + assert_eq!( + WAL_WRITES_AFTER_FAULT.load(Ordering::SeqCst), + 0, + "the warm connection wrote frames after its commit failed" + ); + let second = apply(3); + let failed = section + .events() + .iter() + .any(|event| matches!(event, Event::Failed(_))); + let pending = section + .pending() + .ok() + .map(|pending| pending.iter().map(|edit| edit.id).collect()); + if crash { + std::mem::forget(section); + } else { + section.close().unwrap(); + } + Outcome { + replies: [first, second], + failed, + pending, + } +} + +#[test] +fn commit_fault_child() { + let (Some(root), Some(fault)) = ( + std::env::var_os("COMMIT_FAULT_CHILD"), + std::env::var("COMMIT_FAULT").ok(), + ) else { + return; + }; + let root = Path::new(&root); + let outcome = scenario(root, fault.parse().unwrap(), true); + let ids = |result: &Result| result.as_ref().map_or(0, |id| *id); + std::fs::write( + root.join("outcome"), + format!( + "{} {} {}", + ids(&outcome.replies[0]), + ids(&outcome.replies[1]), + outcome.failed + ), + ) + .unwrap(); + // Ends as a crash would: no SQLite destructor checkpoints or removes the WAL. + std::process::exit(0); +} + +#[test] +fn a_failed_commit_is_decided_by_a_cold_read_and_never_replayed_twice() { + install(); + // (fault, whether the edit reached the cache, whether the cache stays closed) + for (fault, landed, closed) in [ + (SYNC_BEFORE, false, false), + (SYNC_AFTER, false, false), + (WRITE_COMMIT, false, false), + (SYNC_AFTER | KEEP_WAL, true, false), + (SYNC_BEFORE | OPEN, false, true), + ] { + for crash in [false, true] { + let directory = tempfile::tempdir().unwrap(); + let root = directory.path(); + let source = onestore::create_section("Fault.one", "Original", "Fixture").unwrap(); + std::fs::write(root.join("Fault.one"), &source).unwrap(); + drop(Replica::create(root.join("replica.sqlite"), &source).unwrap()); + let (first, second, failed) = if crash { + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "commit_fault_child", "--nocapture"]) + .env("COMMIT_FAULT_CHILD", root) + .env("COMMIT_FAULT", fault.to_string()) + .output() + .unwrap(); + assert!(output.status.success(), "{output:?}"); + let outcome = std::fs::read_to_string(root.join("outcome")).unwrap(); + let mut fields = outcome.split(' '); + let mut id = || Some(fields.next()?.parse::().unwrap()).filter(|id| *id > 0); + let (first, second) = (id(), id()); + (first, second, fields.next() == Some("true")) + } else { + let outcome = scenario(root, fault, false); + let [first, second] = outcome.replies; + if !closed { + let ids = [first.clone().unwrap(), second.clone().unwrap()]; + assert_eq!(&outcome.pending.unwrap()[1..], ids); + } else { + assert!(outcome.pending.is_none()); + } + (first.ok(), second.ok(), outcome.failed) + }; + assert_eq!(failed, !landed, "fault {fault}"); + let cold = Replica::open(root.join("replica.sqlite")).unwrap(); + let colours: Vec<_> = cold + .pending() + .unwrap() + .iter() + .map(|edit| match &edit.edit.ops[..] { + [ + Op::Page { + op: PageOp::Color(Some(colour)), + .. + }, + ] => (edit.id, *colour), + ops => panic!("{ops:?}"), + }) + .collect(); + if closed { + assert_eq!((first, second), (None, None)); + assert_eq!(colours.len(), 1, "fault {fault}"); + } else { + let (first, second) = (first.unwrap(), second.unwrap()); + assert_eq!(&colours[1..], [(first, 2), (second, 3)], "fault {fault}"); + } + } + } +} diff --git a/crates/notebook/tests/support/vfs.rs b/crates/notebook/tests/support/vfs.rs new file mode 100644 index 0000000000000000000000000000000000000000..bbf98abe986aa686b2e6a4623e4b6654c72d3281 --- /dev/null +++ b/crates/notebook/tests/support/vfs.rs @@ -0,0 +1,147 @@ +//! A VFS over the platform's, made the process's default by `install`, that fails what a +//! test arms: once, the WAL's sync or commit frame or the next database open; or every WAL +//! write of the database `FAILING` names, for as long as it names it. +#![allow(dead_code)] + +use rusqlite::ffi; +use std::{ + ffi::{CStr, c_char, c_int, c_void}, + path::PathBuf, + sync::{ + Mutex, Once, OnceLock, + atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering}, + }, +}; + +/// The WAL's sync at commit fails before it reaches the disk. +pub const SYNC_BEFORE: u8 = 1; +/// The WAL's sync at commit reaches the disk and still reports failure. +pub const SYNC_AFTER: u8 = 2; +/// The commit frame's header is never written. +pub const WRITE_COMMIT: u8 = 4; +/// The next open of a database fails, as the cold reopen after a failed commit. +pub const OPEN: u8 = 8; +/// After a fault, the next sync fails, as the checkpoint closing the warm connection makes, +/// which keeps the WAL. +pub const KEEP_WAL: u8 = 16; + +pub static ARMED: AtomicU8 = AtomicU8::new(0); +/// Between a fault and the next database open, as the replica reopens the cache cold. +static FIRED: AtomicBool = AtomicBool::new(false); +pub static WAL_WRITES_AFTER_FAULT: AtomicUsize = AtomicUsize::new(0); +static REAL: OnceLock = OnceLock::new(); +/// The platform's WAL file methods, the shim's copy of them, and the copy failing every write. +static WAL: OnceLock<(usize, usize, usize)> = OnceLock::new(); +/// While set, every write to this database's WAL fails, for WALs opened while it was set. +pub static FAILING: Mutex> = Mutex::new(None); + +fn fire(fault: u8) -> bool { + let fired = ARMED.fetch_and(!fault, Ordering::SeqCst) & fault != 0; + if fired { + FIRED.store(true, Ordering::SeqCst); + } + fired +} + +fn real_wal() -> &'static ffi::sqlite3_io_methods { + unsafe { &*(WAL.get().unwrap().0 as *const ffi::sqlite3_io_methods) } +} + +unsafe extern "C" fn open( + _: *mut ffi::sqlite3_vfs, + name: *const c_char, + file: *mut ffi::sqlite3_file, + flags: c_int, + out: *mut c_int, +) -> c_int { + if flags & ffi::SQLITE_OPEN_MAIN_DB != 0 { + FIRED.store(false, Ordering::SeqCst); + if fire(OPEN) { + FIRED.store(false, Ordering::SeqCst); + unsafe { (*file).pMethods = std::ptr::null() }; + return ffi::SQLITE_CANTOPEN; + } + } + let real = *REAL.get().unwrap() as *mut ffi::sqlite3_vfs; + let opened = unsafe { (*real).xOpen.unwrap()(real, name, file, flags, out) }; + if opened == ffi::SQLITE_OK && flags & ffi::SQLITE_OPEN_WAL != 0 { + let methods = unsafe { (*file).pMethods }; + let &(platform, shim, failing) = WAL.get_or_init(|| { + let mut shim = unsafe { *methods }; + shim.xWrite = Some(write); + shim.xSync = Some(sync); + let mut failing = shim; + failing.xWrite = Some(fail); + let leak = |methods| Box::leak(Box::new(methods)) as *const _ as usize; + (methods as usize, leak(shim), leak(failing)) + }); + assert_eq!(platform, methods as usize); + let name = unsafe { CStr::from_ptr(name) }.to_string_lossy(); + let fails = FAILING + .lock() + .unwrap() + .as_ref() + .is_some_and(|database| name.starts_with(&*database.to_string_lossy())); + let methods = if fails { failing } else { shim }; + unsafe { (*file).pMethods = methods as *const ffi::sqlite3_io_methods }; + } + opened +} + +unsafe extern "C" fn write( + file: *mut ffi::sqlite3_file, + data: *const c_void, + amount: c_int, + offset: ffi::sqlite3_int64, +) -> c_int { + // A frame header names the database size only in the frame that commits. + if amount == 24 + && unsafe { std::slice::from_raw_parts(data.cast::().add(4), 4) } != [0; 4] + && fire(WRITE_COMMIT) + { + return ffi::SQLITE_IOERR_WRITE; + } + if FIRED.load(Ordering::SeqCst) { + WAL_WRITES_AFTER_FAULT.fetch_add(1, Ordering::SeqCst); + } + unsafe { real_wal().xWrite.unwrap()(file, data, amount, offset) } +} + +unsafe extern "C" fn fail( + file: *mut ffi::sqlite3_file, + data: *const c_void, + amount: c_int, + offset: ffi::sqlite3_int64, +) -> c_int { + if FAILING.lock().unwrap().is_some() { + return ffi::SQLITE_IOERR_WRITE; + } + unsafe { write(file, data, amount, offset) } +} + +unsafe extern "C" fn sync(file: *mut ffi::sqlite3_file, flags: c_int) -> c_int { + if fire(SYNC_BEFORE) || FIRED.load(Ordering::SeqCst) && fire(KEEP_WAL) { + return ffi::SQLITE_IOERR_FSYNC; + } + let synced = unsafe { real_wal().xSync.unwrap()(file, flags) }; + if synced == ffi::SQLITE_OK && fire(SYNC_AFTER) { + return ffi::SQLITE_IOERR_FSYNC; + } + synced +} + +pub fn install() { + static INSTALL: Once = Once::new(); + INSTALL.call_once(|| unsafe { + let real = ffi::sqlite3_vfs_find(std::ptr::null()); + REAL.set(real as usize).unwrap(); + let mut shim = *real; + shim.zName = c"commit-fault".as_ptr(); + shim.pNext = std::ptr::null_mut(); + shim.xOpen = Some(open); + assert_eq!( + ffi::sqlite3_vfs_register(Box::leak(Box::new(shim)), 1), + ffi::SQLITE_OK + ); + }); +} diff --git a/crates/snowbound/src/main.rs b/crates/snowbound/src/main.rs index 4fdd0fa24b109e247642dd8f862cbb83b3b209fe..dde03e34d266e7ebeeb17e74ba5e24835ad50e5a 100644 --- a/crates/snowbound/src/main.rs +++ b/crates/snowbound/src/main.rs @@ -4421,7 +4421,7 @@ impl State { eprintln!("Saving failed: {error}"); } } - Event::Failed(error) => eprintln!("Synchronization stopped: {error}"), + Event::Failed(error) => eprintln!("Not saving: {error}"), // Its guests hear of it sooner than the folder's watch would tell them. #[cfg(feature = "live")] Event::Attempt {