authorclover caruso <git@paperclover.net> 2026-10-15 02:38:00-07:00
committerclover caruso <git@paperclover.net> 2026-10-15 04:13:21-07:00
logefa39925fa1ba017f1d50b3cb409f4bd9b11257c
tree277e5df5f5c8dceb343e968bf9910fc522d40c01
parent4abc252b6b516d1b4d937582c1ae2b58693807a2
signature Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU (~clover)

fix(notebook): keep edits until the cache writes them

An edit a session submitted that the cache cannot write for now, through a full disk, an I/O error or a lock, stays applied on the section thread and is written again, in order, each second; Flush, Seal and Rebase answer WouldBlock meanwhile and each failed write reports Event::Failed. An edit the cache could never write is refused alone, as before. A failed COMMIT is decided by a cold reopen of the cache, never by the warm connection, and fails closed when the cache cannot be reopened. iOS keeps a section open while its edits wait. Refs #136 Assisted-by: claude-opus-5.5

11 files changed, 817 insertions(+), 306 deletions(-)

apps/ios/Snowbound/Notebooks.swift+3
...@@ -636,6 +636,8 @@ final class Section {...@@ -636,6 +636,8 @@ final class Section {
636 static let rejected: UInt32 = 4636 static let rejected: UInt32 = 4
637637
638 fileprivate static var open: [Int: Weak<Section>] = [:]638 fileprivate static var open: [Int: Weak<Section>] = [:]
639 /// Sections reporting Not Saving, kept open so that edits waiting in memory stay with them.
640 private static var unsaved: [Int: Section] = [:]
639 private static var tokens = 0641 private static var tokens = 0
640642
641 let notebook: Notebook643 let notebook: Notebook
...@@ -738,6 +740,7 @@ final class Section {...@@ -738,6 +740,7 @@ final class Section {
738 let unlisted = flags & ~Self.listed740 let unlisted = flags & ~Self.listed
739 let unchanged = unlisted == 0 && changed == self.status741 let unchanged = unlisted == 0 && changed == self.status
740 self.status = changed742 self.status = changed
743 Self.unsaved[token] = changed == .notSaving ? self : nil
741 NotificationCenter.default.post(744 NotificationCenter.default.post(
742 name: Self.changed, object: self,745 name: Self.changed, object: self,
743 userInfo: unchanged ? ["flags": unlisted, "unchanged": true] : ["flags": unlisted])746 userInfo: unchanged ? ["flags": unlisted, "unchanged": true] : ["flags": unlisted])
crates/mobile/src/library.rs+2-1
...@@ -1098,7 +1098,7 @@ impl Section {...@@ -1098,7 +1098,7 @@ impl Section {
1098 Status::Offline1098 Status::Offline
1099 }1099 }
1100 Event::Failed(error) => {1100 Event::Failed(error) => {
1101 eprintln!("snowbound: synchronization stopped: {error}");1101 eprintln!("snowbound: not saving: {error}");
1102 Status::NotSaving1102 Status::NotSaving
1103 }1103 }
1104 };1104 };
...@@ -1284,6 +1284,7 @@ impl Library {...@@ -1284,6 +1284,7 @@ impl Library {
1284 let deadline = Instant::now() + limit;1284 let deadline = Instant::now() + limit;
1285 let open = self.open_sections();1285 let open = self.open_sections();
1286 for (_, shared) in &open {1286 for (_, shared) in &open {
1287 shared.section.written().map_err(|_| WAITING)?;
1287 while !shared.section.pending()?.is_empty() {1288 while !shared.section.pending()?.is_empty() {
1288 if Instant::now() >= deadline {1289 if Instant::now() >= deadline {
1289 return Err(WAITING.into());1290 return Err(WAITING.into());
crates/mobile/src/tests.rs+35-118
...@@ -3,6 +3,9 @@ use library::{sb_library_set_offline, sb_library_sync_now};...@@ -3,6 +3,9 @@ use library::{sb_library_set_offline, sb_library_sync_now};
3use notebook::session::SyncState;3use notebook::session::SyncState;
4use std::path::{Path, PathBuf};4use std::path::{Path, PathBuf};
55
6#[path = "../../notebook/tests/support/vfs.rs"]
7mod vfs;
8
6const CORPUS: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/../../corpus");9const CORPUS: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/../../corpus");
710
8fn corpus(path: &str) -> PathBuf {11fn corpus(path: &str) -> PathBuf {
...@@ -3165,14 +3168,9 @@ fn reader_deferred_reload_preserves_rich_composition_after_real_local_write_fail...@@ -3165,14 +3168,9 @@ fn reader_deferred_reload_preserves_rich_composition_after_real_local_write_fail
3165 .section3168 .section
3166 .close()3169 .close()
3167 .unwrap();3170 .unwrap();
3168 let sql = rusqlite::Connection::open(&database).unwrap();3171 // The cache's WAL takes no writes until the test has seen the edits wait.
3169 sql.busy_timeout(Duration::from_secs(1)).unwrap();3172 vfs::install();
3170 sql.execute_batch(3173 *vfs::FAILING.lock().unwrap() = Some(database.canonicalize().unwrap());
3171 "CREATE TRIGGER reader_fault BEFORE INSERT ON edits WHEN NEW.author = 'Fixture'
3172 BEGIN SELECT RAISE(ABORT, 'Disposable Reader write failure'); END;",
3173 )
3174 .unwrap();
3175 sql.close().unwrap();
31763174
3177 let library = Arc::new(Library::open(&file, &cache, true).unwrap());3175 let library = Arc::new(Library::open(&file, &cache, true).unwrap());
3178 sb_library_set_offline(&library, true);3176 sb_library_set_offline(&library, true);
...@@ -3183,7 +3181,6 @@ fn reader_deferred_reload_preserves_rich_composition_after_real_local_write_fail...@@ -3183,7 +3181,6 @@ fn reader_deferred_reload_preserves_rich_composition_after_real_local_write_fail
3183 || {},3181 || {},
3184 )3182 )
3185 .unwrap();3183 .unwrap();
3186 assert_eq!(section.shared.section.identity().unwrap(), identity);
3187 assert_eq!(section.shared.section.pending().unwrap(), seed);3184 assert_eq!(section.shared.section.pending().unwrap(), seed);
3188 let (page, read_only) = section.shared.page(space).unwrap();3185 let (page, read_only) = section.shared.page(space).unwrap();
3189 let mut view = View {3186 let mut view = View {
...@@ -3204,30 +3201,13 @@ fn reader_deferred_reload_preserves_rich_composition_after_real_local_write_fail...@@ -3204,30 +3201,13 @@ fn reader_deferred_reload_preserves_rich_composition_after_real_local_write_fail
3204 );3201 );
3205 assert_ne!(sb_view_format(&view) & 1, 0);3202 assert_ne!(sb_view_format(&view) & 1, 0);
3206 assert!(unsafe { sb_set_marked(&mut view, c"FAILED rich 😀 ".as_ptr(), 15, 15) });3203 assert!(unsafe { sb_set_marked(&mut view, c"FAILED rich 😀 ".as_ptr(), 15, 15) });
3207 let nodes = view.canvas.active().document().nodes().to_vec();
3208 let preview = view.canvas.page.editor.page().unwrap();
3209 section
3210 .shared
3211 .section
3212 .apply(
3213 "Reader peer",
3214 page_edit(space, vec![PageOp::Color(Some(0x0011_2233))]),
3215 )
3216 .unwrap();
3217 section.shared.section.written().unwrap();
3218 let before = section.shared.section.pending().unwrap();
3219 assert_eq!(before.len(), seed.len() + 1);
3220 assert_eq!(&before[..seed.len()], seed.as_slice());
3221 assert!(!sb_view_reload(&mut view, false, Some(&mut anchor)));
3222 assert!(view.reload_pending);
3223 assert!(!sb_view_reload_ready(&view));
3224 if select {3204 if select {
3225 assert!(sb_select(&mut view, 0, 0));3205 assert!(sb_select(&mut view, 0, 0));
3226 } else {3206 } else {
3227 sb_unmark(&mut view);3207 sb_unmark(&mut view);
3228 }3208 }
3229 let history = view.canvas.page.editor.history_depth();3209 assert!(sb_view_set_paper(&mut view, 0x77, 0x88, 0x99, -1));
3230 let selection = view.canvas.page.editor.selection();3210 let preview = view.canvas.page.editor.page().unwrap();
3231 let deadline = Instant::now() + Duration::from_secs(10);3211 let deadline = Instant::now() + Duration::from_secs(10);
3232 let mut flags = 0;3212 let mut flags = 0;
3233 while section.shared.status.load(Ordering::Relaxed) != library::Status::NotSaving as u8 {3213 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...@@ -3237,107 +3217,44 @@ fn reader_deferred_reload_preserves_rich_composition_after_real_local_write_fail
3237 }3217 }
3238 flags |= section.poll();3218 flags |= section.poll();
3239 assert_eq!(flags & 4, 0);3219 assert_eq!(flags & 4, 0);
3240 assert!(view.local_edits.lock().unwrap().failed);3220 assert_eq!(section.shared.section.pending().unwrap(), seed);
3241 assert_eq!(view.canvas.active().document().nodes(), nodes);3221 let (waiting, _) = section.shared.page(space).unwrap();
3242 assert_eq!(view.canvas.page.editor.page().unwrap(), preview);3222 assert!(canvas::search::page_text(&waiting).contains("FAILED rich 😀"));
3243 assert!(view.canvas.active().shown_text().contains("FAILED rich 😀"));3223 assert_eq!(waiting.color, Some(0x0099_8877));
3244 assert!(view.canvas.edit().unwrap().is_none());3224 assert_eq!(view.local_edits.lock().unwrap().pending, 2);
3245 section.shared.section.written().unwrap();
3246 assert_eq!(section.shared.section.pending().unwrap(), before);
3247 assert!(!sb_view_reload(&mut view, false, Some(&mut anchor)));
3248 assert!(!sb_view_reload_ready(&view));
3249 assert!(view.reload_pending);
3250 assert_eq!(view.canvas.page.editor.page().unwrap(), preview);
3251 assert_eq!(view.canvas.page.editor.history_depth(), history);
3252 assert_eq!(view.canvas.page.editor.selection(), selection);
3253 assert_eq!(std::fs::read(&file).unwrap(), original);
3254
3255 section
3256 .shared
3257 .section
3258 .apply(
3259 "Reader peer",
3260 page_edit(space, vec![PageOp::Color(Some(0x0044_5566))]),
3261 )
3262 .unwrap();
3263 section.shared.section.written().unwrap();
3264 let _ = section.poll();
3265 assert!(!sb_view_reload_ready(&view));
3266 assert!(!sb_view_reload(&mut view, false, None));
3267 assert_eq!(view.canvas.page.editor.page().unwrap(), preview);
3268 unsafe { library::sb_section_set_author(&section, c"Reader peer".as_ptr()) };
3269 assert!(sb_view_set_paper(&mut view, 0x77, 0x88, 0x99, -1));
3270 section.shared.section.written().unwrap();
3271 let successful_later = section.shared.section.pending().unwrap();
3272 assert_eq!(successful_later.len(), before.len() + 2);
3273 assert_eq!(successful_later.last().unwrap().author, "Reader peer");
3274 assert!(view.local_edits.lock().unwrap().failed);
3275 assert!(!sb_view_reload_ready(&view));
3276 assert!(view.canvas.active().shown_text().contains("FAILED rich 😀"));
3277 let outstanding = ApplyReply::new(&view.local_edits);
3278 let (returned, returned_reply) = std::sync::mpsc::channel();
3279 section
3280 .shared
3281 .section
3282 .apply_with(
3283 "Fixture",
3284 page_edit(space, vec![PageOp::Color(Some(0x0010_2030))]),
3285 move |result| returned.send((outstanding, result)).unwrap(),
3286 )
3287 .unwrap();
3288 let (outstanding, result) = returned_reply
3289 .recv_timeout(Duration::from_secs(10))
3290 .unwrap();
3291 assert!(result.is_err());
3292 assert_eq!(section.shared.section.pending().unwrap(), successful_later);
3293 sb_view_reload(&mut view, true, None);
3294 assert!(!view.canvas.active().shown_text().contains("FAILED rich 😀"));
3295 assert_eq!(view.local_edits.lock().unwrap().pending, 1);
3296 assert!(!view.local_edits.lock().unwrap().failed);3225 assert!(!view.local_edits.lock().unwrap().failed);
3297 sb_view_reload(&mut view, false, None);3226 assert!(library.close(Duration::from_millis(100)).is_err());
3298 assert!(!sb_view_reload_ready(&view));3227 *vfs::FAILING.lock().unwrap() = None;
3299 outstanding.complete(result);3228
3300 assert!(view.local_edits.lock().unwrap().failed);3229 let deadline = Instant::now() + Duration::from_secs(10);
3301 assert!(!sb_view_reload_ready(&view));3230 while view.local_edits.lock().unwrap().pending > 0 {
3302 sb_view_reload(&mut view, true, None);3231 assert!(Instant::now() < deadline);
3232 std::thread::sleep(Duration::from_millis(10));
3233 }
3303 assert!(!view.local_edits.lock().unwrap().failed);3234 assert!(!view.local_edits.lock().unwrap().failed);
3304 assert_eq!(view.local_edits.lock().unwrap().pending, 0);3235 let pending = section.shared.section.pending().unwrap();
3305 assert!(!view.reload_pending);3236 assert_eq!(pending.len(), seed.len() + 2);
3306 let _ = section.poll();3237 assert_eq!(&pending[..seed.len()], seed.as_slice());
3307 assert!(sb_view_set_paper(&mut view, 0x11, 0x22, 0x33, -1));
3308 section.shared.section.written().unwrap();
3309 section
3310 .shared
3311 .section
3312 .apply(
3313 "Reader peer",
3314 page_edit(space, vec![PageOp::Color(Some(0x0012_3456))]),
3315 )
3316 .unwrap();
3317 section.shared.section.written().unwrap();
3318 sb_view_reload(&mut view, false, None);3238 sb_view_reload(&mut view, false, None);
3319 assert!(!view.reload_pending);3239 assert!(!view.reload_pending);
3320 assert!(!view.local_edits.lock().unwrap().failed);3240 assert_eq!(view.canvas.page.editor.page().unwrap(), preview);
3321 assert_eq!(
3322 view.canvas.page.editor.page().unwrap(),
3323 section.shared.page(space).unwrap().0
3324 );
3325 assert_eq!(std::fs::read(&file).unwrap(), original);
3326 let preserved = section.shared.section.pending().unwrap();
3327 let shared = Arc::clone(&section.shared);3241 let shared = Arc::clone(&section.shared);
3328 drop((view, section, library));3242 drop((view, section, library));
3329 Arc::into_inner(shared)3243 Arc::into_inner(shared)
3330 .expect("No other failed writer")3244 .expect("No other writer")
3331 .section3245 .section
3332 .close()3246 .close()
3333 .unwrap();3247 .unwrap();
3334 let sql = rusqlite::Connection::open(&database).unwrap();
3335 sql.busy_timeout(Duration::from_secs(1)).unwrap();
3336 sql.execute_batch("DROP TRIGGER reader_fault").unwrap();
3337 sql.close().unwrap();
3338 let cold = notebook::Replica::open(&database).unwrap();3248 let cold = notebook::Replica::open(&database).unwrap();
3339 assert_eq!(cold.identity().unwrap(), identity);3249 assert_eq!(cold.identity().unwrap(), identity);
3340 assert_eq!(cold.pending().unwrap(), preserved);3250 assert_eq!(cold.pending().unwrap(), pending);
3341 assert!(!canvas::search::page_text(&cold.page(space).unwrap()).contains("FAILED rich 😀"));3251 let page = cold.page(space).unwrap();
3252 assert_eq!(
3253 canvas::search::page_text(&page)
3254 .matches("FAILED rich 😀")
3255 .count(),
3256 1
3257 );
3258 assert_eq!(page.color, Some(0x0099_8877));
3342 }3259 }
3343}3260}
crates/notebook/README.md+5-2
...@@ -44,8 +44,11 @@ it was, then hands its requests over; edits sent meanwhile apply to the rebuilt...@@ -44,8 +44,11 @@ it was, then hands its requests over; edits sent meanwhile apply to the rebuilt
4444
45Share one `Replica` between threads. Keep the cache on a local filesystem: the45Share one `Replica` between threads. Keep the cache on a local filesystem: the
46connection holds exclusive ownership between transactions, and a second open fails46connection holds exclusive ownership between transactions, and a second open fails
47busy. No network wait occurs in a local edit. After a database error the section47busy. No network wait occurs in a local edit. An edit a session submitted that the cache
48thread rereads the queue.48cannot write for now (a full disk, an I/O error) stays applied in memory and is written
49again, in order, each second; meanwhile `written` returns `WouldBlock`. After a commit
50fails, a fresh connection decides whether it landed. An edit the cache could never write
51is refused alone, and the section thread rereads the queue.
4952
50## Sessions53## Sessions
5154
crates/notebook/src/lib.rs+13-7
...@@ -302,26 +302,32 @@ impl Replica {...@@ -302,26 +302,32 @@ impl Replica {
302302
303 /// Applies an edit, awaiting its durable local queue receipt without blocking browser reads.303 /// Applies an edit, awaiting its durable local queue receipt without blocking browser reads.
304 pub async fn apply_async(&self, author: &str, edit: Edit) -> Result<u64> {304 pub async fn apply_async(&self, author: &str, edit: Edit) -> Result<u64> {
305 self.ask_async(|reply| working::Request::Apply {305 self.ask_async(|reply| {
306 author: author.to_owned(),306 working::Request::Apply(working::Accepted {
307 edit,307 author: author.to_owned(),
308 reply,308 edit,
309 reply,
310 waiting: None,
311 })
309 })312 })
310 .await313 .await
311 }314 }
312315
313 /// `apply` without waiting: `reply` runs once the edit is durable or refused.316 /// `apply` without waiting: `reply` runs once the edit is durable or refused. While the
317 /// cache refuses to write it, the edit waits in memory and each failed write tells `waiting`.
314 pub(crate) fn submit(318 pub(crate) fn submit(
315 &self,319 &self,
316 author: &str,320 author: &str,
317 edit: Edit,321 edit: Edit,
318 reply: working::Reply<u64>,322 reply: working::Reply<u64>,
323 waiting: working::Notice,
319 ) -> Result<()> {324 ) -> Result<()> {
320 self.send(working::Request::Apply {325 self.send(working::Request::Apply(working::Accepted {
321 author: author.to_owned(),326 author: author.to_owned(),
322 edit,327 edit,
323 reply,328 reply,
324 })329 waiting: Some(waiting),
330 }))
325 }331 }
326332
327 /// The page in `space` as the queued edits leave it; O(page), for opening and reloading.333 /// The page in `space` as the queued edits leave it; O(page), for opening and reloading.
crates/notebook/src/session.rs+10-1
...@@ -2136,7 +2136,7 @@ pub enum Event {...@@ -2136,7 +2136,7 @@ pub enum Event {
2136 Attempt { id: u64, status: EditStatus },2136 Attempt { id: u64, status: EditStatus },
2137 /// The section file could not be reached; the replica keeps its state.2137 /// The section file could not be reached; the replica keeps its state.
2138 Unreachable(io::Error),2138 Unreachable(io::Error),
2139 /// The replica itself failed; the worker has stopped.2139 /// The replica itself failed and the worker stopped, or edits wait in memory for the cache.
2140 Failed(String),2140 Failed(String),
2141}2141}
21422142
...@@ -2482,6 +2482,14 @@ impl Section {...@@ -2482,6 +2482,14 @@ impl Section {
2482 reply: impl FnOnce(Result<u64>) + Send + 'static,2482 reply: impl FnOnce(Result<u64>) + Send + 'static,
2483 ) -> Result<()> {2483 ) -> Result<()> {
2484 let (sender, notify) = (self.sender.clone(), Arc::clone(&self.notify));2484 let (sender, notify) = (self.sender.clone(), Arc::clone(&self.notify));
2485 let waiting: crate::working::Notice = {
2486 let (sender, notify) = (sender.clone(), Arc::clone(&notify));
2487 Box::new(move |error| {
2488 if sender.send(Event::Failed(error.to_owned())).is_ok() {
2489 notify();
2490 }
2491 })
2492 };
2485 let root = self.replica.root;2493 let root = self.replica.root;
2486 let spaces = crate::queue::spaces(&edit, root);2494 let spaces = crate::queue::spaces(&edit, root);
2487 self.replica.submit(2495 self.replica.submit(
...@@ -2505,6 +2513,7 @@ impl Section {...@@ -2505,6 +2513,7 @@ impl Section {
2505 notify();2513 notify();
2506 }2514 }
2507 }),2515 }),
2516 waiting,
2508 )2517 )
2509 }2518 }
25102519
crates/notebook/src/session/tests.rs+183
...@@ -670,3 +670,186 @@ fn associated_apply_reply_precedes_success_wake_without_a_publication_event() {...@@ -670,3 +670,186 @@ fn associated_apply_reply_precedes_success_wake_without_a_publication_event() {
670 section.close().unwrap();670 section.close().unwrap();
671 assert_eq!(fs::read(file).unwrap(), source);671 assert_eq!(fs::read(file).unwrap(), source);
672}672}
673
674#[test]
675fn edits_the_cache_refuses_wait_in_order_and_land_once() {
676 use std::time::Instant;
677
678 let temporary = tempfile::tempdir().unwrap();
679 let file = temporary.path().join("Waiting.one");
680 let cache = temporary.path().join("waiting.sqlite");
681 let source = onestore::create_section("Waiting.one", "Original", "Fixture").unwrap();
682 fs::write(&file, &source).unwrap();
683 let replica = Replica::create(&cache, &source).unwrap();
684 let section = Section::resume(&file, replica, true, || {}).unwrap();
685 let replica = Arc::clone(section.replica());
686 // Queued before the fault, so the worker asks to seal while edits wait.
687 let seed = replica
688 .apply(
689 "Fixture",
690 Edit {
691 at: crate::now(),
692 ops: vec![Op::Section(SectionOp::Create(
693 PageCreation::new(None, Some("Seed"), "Fixture").unwrap(),
694 ))],
695 },
696 )
697 .unwrap();
698 // The cache is full, until it may grow again; the first edit needs room.
699 let full = |pages: Option<i64>| {
700 let connection = replica.lock().unwrap();
701 let pages = pages.unwrap_or_else(|| {
702 connection
703 .pragma_query_value(None, "page_count", |row| row.get(0))
704 .unwrap()
705 });
706 connection
707 .pragma_update(None, "max_page_count", pages)
708 .unwrap();
709 };
710 full(None);
711 let title = "Waiting page ".repeat(1000);
712 let creation = PageCreation::new(None, Some(&title), "Fixture").unwrap();
713 let space = creation.space();
714 // The second edit colours the page the first creates.
715 let edits = [
716 Op::Section(SectionOp::Create(creation)),
717 Op::Page {
718 space,
719 op: onestore::op::PageOp::Color(Some(0x00ab_cdef)),
720 },
721 ]
722 .map(|op| Edit {
723 at: crate::now(),
724 ops: vec![op],
725 });
726 let (replied, replies) = mpsc::channel();
727 for edit in edits.clone() {
728 let replied = replied.clone();
729 section
730 .apply_with("Fixture", edit, move |result| {
731 replied
732 .send(result.map_err(|error| error.to_string()))
733 .unwrap();
734 })
735 .unwrap();
736 }
737 let deadline = Instant::now() + Duration::from_secs(10);
738 while !section
739 .events()
740 .iter()
741 .any(|event| matches!(event, Event::Failed(_)))
742 {
743 assert!(Instant::now() < deadline);
744 std::thread::sleep(Duration::from_millis(10));
745 }
746 let waiting = |section: &Section| {
747 assert!(replies.try_recv().is_err());
748 let pending = section.pending().unwrap();
749 assert_eq!(
750 pending.iter().map(|edit| edit.id).collect::<Vec<_>>(),
751 [seed]
752 );
753 let page = section.page(space).unwrap();
754 assert_eq!((&page.title, page.color), (&title, Some(0x00ab_cdef)));
755 assert!(matches!(
756 replica.ask(|reply| crate::working::Request::Seal { reply }),
757 Err(Error::Io(error)) if error.kind() == io::ErrorKind::WouldBlock
758 ));
759 assert!(matches!(
760 section.written(),
761 Err(Error::Io(error)) if error.kind() == io::ErrorKind::WouldBlock
762 ));
763 };
764 waiting(&section);
765 // A caller blocking on its edit still learns of the failure, and the waiting edits stay.
766 let blocked = Edit {
767 at: crate::now(),
768 ops: vec![Op::Page {
769 space,
770 op: onestore::op::PageOp::Color(Some(0x0012_3456)),
771 }],
772 };
773 assert!(replica.apply("Fixture", blocked).is_err());
774 waiting(&section);
775 section.wake();
776 std::thread::sleep(Duration::from_millis(200));
777 waiting(&section);
778
779 full(Some(1 << 30));
780 let ids = [(); 2].map(|()| {
781 replies
782 .recv_timeout(Duration::from_secs(10))
783 .unwrap()
784 .unwrap()
785 });
786 let pending = section.pending().unwrap();
787 assert_eq!(
788 pending[1..]
789 .iter()
790 .map(|edit| (edit.id, &edit.edit))
791 .collect::<Vec<_>>(),
792 [(ids[0], &edits[0]), (ids[1], &edits[1])]
793 );
794
795 // An edit that could never be written is refused alone; the next one lands.
796 replica
797 .lock()
798 .unwrap()
799 .execute_batch(
800 "CREATE TEMP TRIGGER refuse BEFORE INSERT ON main.edits WHEN NEW.author = 'Refused'
801 BEGIN SELECT RAISE(ABORT, 'Disposable refusal'); END;",
802 )
803 .unwrap();
804 for (author, colour) in [("Refused", 0x0012_3456), ("Fixture", 0x0065_4321)] {
805 let replied = replied.clone();
806 section
807 .apply_with(
808 author,
809 Edit {
810 at: crate::now(),
811 ops: vec![Op::Page {
812 space,
813 op: onestore::op::PageOp::Color(Some(colour)),
814 }],
815 },
816 move |result| {
817 replied
818 .send(result.map_err(|error| error.to_string()))
819 .unwrap();
820 },
821 )
822 .unwrap();
823 }
824 assert!(
825 replies
826 .recv_timeout(Duration::from_secs(10))
827 .unwrap()
828 .is_err()
829 );
830 let last = replies
831 .recv_timeout(Duration::from_secs(10))
832 .unwrap()
833 .unwrap();
834 section.written().unwrap();
835 assert_eq!(section.pending().unwrap().last().unwrap().id, last);
836 assert_eq!(section.page(space).unwrap().color, Some(0x0065_4321));
837 section.wake();
838 let deadline = Instant::now() + Duration::from_secs(10);
839 while !section.events().iter().any(|event| {
840 matches!(event, Event::Attempt { id, status: EditStatus::Published { .. } } if *id == last)
841 }) {
842 assert!(Instant::now() < deadline);
843 std::thread::sleep(Duration::from_millis(10));
844 }
845 section.close().unwrap();
846 drop(replica);
847 let pages = stored_pages(&fs::read(&file).unwrap()).unwrap();
848 let waited: Vec<_> = pages
849 .iter()
850 .filter(|page| page.page.title == title)
851 .collect();
852 assert_eq!(waited.len(), 1);
853 assert_eq!(waited[0].page.color, Some(0x0065_4321));
854 assert!(Replica::open(&cache).unwrap().pending().unwrap().is_empty());
855}
crates/notebook/src/working.rs+241-176
...@@ -15,17 +15,15 @@ use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};...@@ -15,17 +15,15 @@ use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
15use std::{15use std::{
16 collections::{BTreeMap, BTreeSet, VecDeque},16 collections::{BTreeMap, BTreeSet, VecDeque},
17 io,17 io,
18 path::PathBuf,
18 sync::{Arc, Mutex, Weak, mpsc},19 sync::{Arc, Mutex, Weak, mpsc},
19};20};
2021
21pub(crate) type Reply<T> = Box<dyn FnOnce(Result<T>) + Send>;22pub(crate) type Reply<T> = Box<dyn FnOnce(Result<T>) + Send>;
23pub(crate) type Notice = Box<dyn Fn(&str) + Send>;
2224
23pub(crate) enum Request {25pub(crate) enum Request {
24 Apply {26 Apply(Accepted),
25 author: String,
26 edit: Edit,
27 reply: Reply<u64>,
28 },
29 Page {27 Page {
30 space: ExGuid,28 space: ExGuid,
31 reply: Reply<Page>,29 reply: Reply<Page>,
...@@ -82,10 +80,10 @@ pub(crate) struct Sealed {...@@ -82,10 +80,10 @@ pub(crate) struct Sealed {
82}80}
8381
84impl Request {82impl Request {
85 fn fail(self, message: &str) {83 fn fail(self, kind: io::ErrorKind, message: &str) {
86 let error = || io::Error::other(message.to_owned()).into();84 let error = || io::Error::new(kind, message.to_owned()).into();
87 match self {85 match self {
88 Self::Apply { reply, .. } => reply(Err(error())),86 Self::Apply(Accepted { reply, .. }) => reply(Err(error())),
89 Self::Page { reply, .. } => reply(Err(error())),87 Self::Page { reply, .. } => reply(Err(error())),
90 Self::Pages { reply } => reply(Err(error())),88 Self::Pages { reply } => reply(Err(error())),
91 Self::Conflicts { reply } => reply(Err(error())),89 Self::Conflicts { reply } => reply(Err(error())),
...@@ -288,19 +286,33 @@ mod serving {...@@ -288,19 +286,33 @@ mod serving {
288 }286 }
289}287}
290288
291/// The next request, or none once the replica dropped.289/// The next request; a second passes without one while `retry`.
292#[cfg(not(target_arch = "wasm32"))]290#[cfg(not(target_arch = "wasm32"))]
293async fn receive(requests: &mpsc::Receiver<Request>) -> Option<Request> {291async fn receive(
294 requests.recv().ok()292 requests: &mpsc::Receiver<Request>,
293 retry: bool,
294) -> std::result::Result<Request, mpsc::RecvTimeoutError> {
295 if retry {
296 requests.recv_timeout(std::time::Duration::from_secs(1))
297 } else {
298 requests
299 .recv()
300 .map_err(|_| mpsc::RecvTimeoutError::Disconnected)
301 }
295}302}
296303
297/// The next request, waiting for `serving::serve` where none is queued, or none once the304/// The next request, waiting for `serving::serve` where none is queued; requests alone
298/// replica dropped.305/// retry unwritten edits.
299#[cfg(target_arch = "wasm32")]306#[cfg(target_arch = "wasm32")]
300async fn receive(requests: &mpsc::Receiver<Request>) -> Option<Request> {307async fn receive(
308 requests: &mpsc::Receiver<Request>,
309 _retry: bool,
310) -> std::result::Result<Request, mpsc::RecvTimeoutError> {
301 std::future::poll_fn(|_| match requests.try_recv() {311 std::future::poll_fn(|_| match requests.try_recv() {
302 Ok(request) => std::task::Poll::Ready(Some(request)),312 Ok(request) => std::task::Poll::Ready(Ok(request)),
303 Err(mpsc::TryRecvError::Disconnected) => std::task::Poll::Ready(None),313 Err(mpsc::TryRecvError::Disconnected) => {
314 std::task::Poll::Ready(Err(mpsc::RecvTimeoutError::Disconnected))
315 }
304 Err(mpsc::TryRecvError::Empty) => std::task::Poll::Pending,316 Err(mpsc::TryRecvError::Empty) => std::task::Poll::Pending,
305 })317 })
306 .await318 .await
...@@ -386,7 +398,7 @@ async fn run(...@@ -386,7 +398,7 @@ async fn run(
386fn fail(error: crate::Error, backlog: VecDeque<Request>, requests: mpsc::Receiver<Request>) {398fn fail(error: crate::Error, backlog: VecDeque<Request>, requests: mpsc::Receiver<Request>) {
387 let message = error.to_string();399 let message = error.to_string();
388 for request in backlog.into_iter().chain(requests.iter()) {400 for request in backlog.into_iter().chain(requests.iter()) {
389 request.fail(&message);401 request.fail(io::ErrorKind::Other, &message);
390 }402 }
391}403}
392404
...@@ -462,11 +474,13 @@ fn inline(shared: &Thread, job: Job) -> bool {...@@ -462,11 +474,13 @@ fn inline(shared: &Thread, job: Job) -> bool {
462 true474 true
463}475}
464476
465/// An edit applied to the section and not yet written, with who waits for it.477/// An edit to apply, with who waits for it.
466struct Accepted {478pub(crate) struct Accepted {
467 author: String,479 pub author: String,
468 edit: Edit,480 pub edit: Edit,
469 reply: Reply<u64>,481 pub reply: Reply<u64>,
482 /// Set where nobody blocks on the reply: a failed write keeps the edit and reports here.
483 pub waiting: Option<Notice>,
470}484}
471485
472struct Working<'a> {486struct Working<'a> {
...@@ -475,6 +489,9 @@ struct Working<'a> {...@@ -475,6 +489,9 @@ struct Working<'a> {
475 open: Option<i64>,489 open: Option<i64>,
476 /// Spaces the open batch's edits change.490 /// Spaces the open batch's edits change.
477 touched: BTreeSet<ExGuid>,491 touched: BTreeSet<ExGuid>,
492 /// Applied to the section and not yet in the queue, so the section is the queue replayed
493 /// with these after it; written in order before anything seals, rebases or rereads.
494 unwritten: Vec<Accepted>,
478}495}
479496
480impl<'a> Working<'a> {497impl<'a> Working<'a> {
...@@ -484,6 +501,7 @@ impl<'a> Working<'a> {...@@ -484,6 +501,7 @@ impl<'a> Working<'a> {
484 section,501 section,
485 open,502 open,
486 touched,503 touched,
504 unwritten: Vec::new(),
487 })505 })
488 }506 }
489507
...@@ -498,20 +516,24 @@ impl<'a> Working<'a> {...@@ -498,20 +516,24 @@ impl<'a> Working<'a> {
498 // other request waits for the rebuilt section.516 // other request waits for the rebuilt section.
499 let mut held: Option<VecDeque<Request>> = None;517 let mut held: Option<VecDeque<Request>> = None;
500 loop {518 loop {
501 let first = match backlog.pop_front() {519 let mut burst = std::mem::take(backlog);
502 Some(request) => request,520 // Only a burst that took an edit, or a retry's tick, writes at its end: reads never.
503 None => match receive(requests).await {521 let mut write = false;
504 Some(request) => request,522 if burst.is_empty() {
505 None => return Next::Stop,523 match receive(requests, !self.unwritten.is_empty()).await {
506 },524 Ok(request) => burst.push_back(request),
507 };525 Err(mpsc::RecvTimeoutError::Timeout) => write = true,
508 let mut burst: VecDeque<Request> = VecDeque::from([first]);526 Err(mpsc::RecvTimeoutError::Disconnected) => {
509 burst.extend(backlog.drain(..));527 let _ = self.flush(shared);
528 return Next::Stop;
529 }
530 }
531 }
510 burst.extend(requests.try_iter());532 burst.extend(requests.try_iter());
511 let mut accepted = Vec::new();
512 // Requests that write the queue beyond the burst's edits run after its reads.533 // Requests that write the queue beyond the burst's edits run after its reads.
513 let mut later = VecDeque::new();534 let mut later = VecDeque::new();
514 while let Some(request) = burst.pop_front().or_else(|| later.pop_front()) {535 let mut reopen = false;
536 while !reopen && let Some(request) = burst.pop_front().or_else(|| later.pop_front()) {
515 if let Some(waiting) = &mut held {537 if let Some(waiting) = &mut held {
516 match request {538 match request {
517 Request::Page { space, reply } => {539 Request::Page { space, reply } => {
...@@ -542,15 +564,29 @@ impl<'a> Working<'a> {...@@ -542,15 +564,29 @@ impl<'a> Working<'a> {
542 }564 }
543 continue;565 continue;
544 }566 }
545 let reopen = match request {567 if matches!(request, Request::Flush { .. } | Request::Seal { .. })
546 Request::Apply {568 && !burst.is_empty()
547 author,569 {
548 edit,570 later.push_back(request);
549 reply,571 continue;
550 } => self572 }
551 .accept(author, edit, reply, &mut accepted, shared)573 if matches!(
552 .err()574 request,
553 .unwrap_or(false),575 Request::Flush { .. } | Request::Seal { .. } | Request::Rebase { .. }
576 ) {
577 let failed = self.flush(shared).is_err();
578 if failed || !self.unwritten.is_empty() {
579 // The worker retries later; the edits wait here meanwhile.
580 request.fail(io::ErrorKind::WouldBlock, "Edits wait to be written");
581 reopen = failed;
582 continue;
583 }
584 }
585 reopen = match request {
586 Request::Apply(accepted) => {
587 write = true;
588 self.accept(accepted)
589 }
554 Request::Page { space, reply } => {590 Request::Page { space, reply } => {
555 reply(self.section.page(space).map_err(Into::into));591 reply(self.section.page(space).map_err(Into::into));
556 false592 false
...@@ -575,86 +611,62 @@ impl<'a> Working<'a> {...@@ -575,86 +611,62 @@ impl<'a> Working<'a> {
575 reply(self.section.version(space, version).map_err(Into::into));611 reply(self.section.version(space, version).map_err(Into::into));
576 false612 false
577 }613 }
578 request @ (Request::Flush { .. } | Request::Seal { .. })
579 if !burst.is_empty() =>
580 {
581 later.push_back(request);
582 false
583 }
584 Request::Flush { reply, durable } => {614 Request::Flush { reply, durable } => {
585 let written = self.flush(shared, &mut accepted);615 #[cfg(target_arch = "wasm32")]
586 if written {616 if durable {
587 #[cfg(target_arch = "wasm32")]617 shared.receipt(vec![reply], false);
588 if durable {
589 shared.receipt(vec![reply], false);
590 } else {
591 reply(Ok(()));
592 }
593 #[cfg(not(target_arch = "wasm32"))]
594 {
595 let _ = durable;
596 reply(Ok(()));
597 }
598 } else {618 } else {
599 reply(Err(619 reply(Ok(()));
600 io::Error::other("The queue could not be written").into()620 }
601 ));621 #[cfg(not(target_arch = "wasm32"))]
622 {
623 let _ = durable;
624 reply(Ok(()));
602 }625 }
603 !written626 false
604 }627 }
605 Request::Seal { reply } => {628 Request::Seal { reply } => {
606 if !self.flush(shared, &mut accepted) {629 let sealed = self.seal(connection);
607 reply(Err(630 let failed = sealed.is_err();
608 io::Error::other("The queue could not be written").into()631 reply(sealed);
609 ));632 failed
610 true
611 } else {
612 let sealed = self.seal(connection);
613 let failed = sealed.is_err();
614 reply(sealed);
615 failed
616 }
617 }633 }
618 Request::Rebase { image, reply } => {634 Request::Rebase { image, reply } => {
619 if self.flush(shared, &mut accepted) {635 #[cfg(target_arch = "wasm32")]
620 #[cfg(target_arch = "wasm32")]636 let reopen = inline(shared, Job::Rebase { image, reply });
621 let reopen = inline(shared, Job::Rebase { image, reply });637 #[cfg(not(target_arch = "wasm32"))]
622 #[cfg(not(target_arch = "wasm32"))]638 let reopen = {
623 let reopen = {639 held = self
624 held = self640 .rebuild(shared, Job::Rebase { image, reply })
625 .rebuild(shared, Job::Rebase { image, reply })641 .then(VecDeque::new);
626 .then(VecDeque::new);642 false
627 false643 };
628 };644 reopen
629 reopen
630 } else {
631 reply(Err(
632 io::Error::other("The queue could not be written").into()
633 ));
634 true
635 }
636 }645 }
637 Request::Reopen { reply } => {646 Request::Reopen { reply } => {
638 self.flush(shared, &mut accepted);647 let _ = self.flush(shared);
639 #[cfg(target_arch = "wasm32")]648 #[cfg(target_arch = "wasm32")]
640 let reopen = inline(shared, Job::Reopen { reply });649 let reopen = inline(shared, Job::Reopen { reply });
641 #[cfg(not(target_arch = "wasm32"))]650 #[cfg(not(target_arch = "wasm32"))]
642 let reopen = {651 let reopen = {
643 held = self652 held = self
644 .rebuild(shared, Job::Reopen { reply })653 .rebuild(shared, Job::Reopen { reply })
645 .then(VecDeque::new);654 .then(|| self.unwritten.drain(..).map(Request::Apply).collect());
646 false655 false
647 };656 };
648 reopen657 reopen
649 }658 }
650 Request::Handover(_) | Request::Resume => false,659 Request::Handover(_) | Request::Resume => false,
651 };660 };
652 if reopen {
653 backlog.extend(burst.drain(..).chain(later.drain(..)));
654 return Next::Reopen;
655 }
656 }661 }
657 if !self.flush(shared, &mut accepted) {662 if reopen || write && self.flush(shared).is_err() {
663 backlog.extend(
664 self.unwritten
665 .drain(..)
666 .map(Request::Apply)
667 .chain(burst.drain(..))
668 .chain(later.drain(..)),
669 );
658 return Next::Reopen;670 return Next::Reopen;
659 }671 }
660 }672 }
...@@ -686,101 +698,120 @@ impl<'a> Working<'a> {...@@ -686,101 +698,120 @@ impl<'a> Working<'a> {
686 }698 }
687699
688 /// Applies an edit, keeping it to be written with the burst. A refused edit is answered700 /// Applies an edit, keeping it to be written with the burst. A refused edit is answered
689 /// here; `Err(true)` when it failed part way and the section must be reread.701 /// here; true when it failed part way and the section must be reread.
690 fn accept(702 fn accept(&mut self, accepted: Accepted) -> bool {
691 &mut self,703 match self.section.apply(&accepted.author, &accepted.edit) {
692 author: String,
693 edit: Edit,
694 reply: Reply<u64>,
695 accepted: &mut Vec<Accepted>,
696 shared: &Arc<Thread>,
697 ) -> std::result::Result<(), bool> {
698 match self.section.apply(&author, &edit) {
699 Ok(()) => {704 Ok(()) => {
700 self.touched705 self.touched
701 .extend(queue::spaces(&edit, self.section.root()));706 .extend(queue::spaces(&accepted.edit, self.section.root()));
702 accepted.push(Accepted {707 self.unwritten.push(accepted);
703 author,708 false
704 edit,
705 reply,
706 });
707 Ok(())
708 }709 }
709 Err(error) => {710 Err(error) => {
710 let broken = matches!(error, OpError::Failed(_));711 let broken = matches!(error, OpError::Failed(_));
711 if broken {712 (accepted.reply)(Err(error.into()));
712 self.flush(shared, accepted);713 broken
713 }
714 reply(Err(error.into()));
715 Err(broken)
716 }714 }
717 }715 }
718 }716 }
719717
720 /// Writes the accepted edits in one transaction, then answers their senders; false718 /// Writes the unwritten edits in one transaction, then answers their senders. A write
721 /// when the write failed and the section holds edits the queue lacks.719 /// that may succeed later keeps the edits with a notice, telling the first; one that never
722 fn flush(&mut self, shared: &Arc<Thread>, accepted: &mut Vec<Accepted>) -> bool {720 /// will refuses the edit it failed on. An error means the section holds an edit the queue
723 if accepted.is_empty() {721 /// never will, or the cache closed, and must be reread.
724 return true;722 fn flush(&mut self, shared: &Arc<Thread>) -> Result<()> {
723 if self.unwritten.is_empty() {
724 return Ok(());
725 }725 }
726 let written = (|| -> Result<Vec<u64>> {726 let mut connection = lock(&shared.connection)?;
727 let mut connection = lock(&shared.connection)?;727 let mut ids = Vec::new();
728 let transaction =728 let committed = connection
729 connection.transaction_with_behavior(TransactionBehavior::Immediate)?;729 .transaction_with_behavior(TransactionBehavior::Immediate)
730 let batch = match self.open {730 .map_err(crate::Error::from)
731 Some(batch) => batch,731 .and_then(|transaction| {
732 None => {732 let batch = match self.open {
733 transaction.execute("INSERT INTO batches DEFAULT VALUES", [])?;733 Some(batch) => batch,
734 transaction.last_insert_rowid()734 None => {
735 }735 transaction.execute("INSERT INTO batches DEFAULT VALUES", [])?;
736 };736 transaction.last_insert_rowid()
737 let mut ids = Vec::new();
738 for edit in accepted.iter() {
739 ids.push(queue::insert(
740 &transaction,
741 shared.key.as_ref(),
742 None,
743 batch,
744 &edit.author,
745 &edit.edit,
746 )?);
747 }
748 transaction.commit()?;
749 self.open = Some(batch);
750 Ok(ids)
751 })();
752 let ok = written.is_ok();
753 match written {
754 Ok(ids) => {
755 #[cfg(target_arch = "wasm32")]
756 shared.receipt(
757 accepted
758 .drain(..)
759 .zip(ids)
760 .map(|(edit, id)| {
761 let reply: Reply<()> =
762 Box::new(move |result| (edit.reply)(result.map(|()| id)));
763 reply
764 })
765 .collect(),
766 true,
767 );
768 #[cfg(not(target_arch = "wasm32"))]
769 {
770 for (edit, id) in accepted.drain(..).zip(ids) {
771 (edit.reply)(Ok(id));
772 }737 }
773 crate::edited(&shared.worker);738 };
739 for edit in &self.unwritten {
740 ids.push(queue::insert(
741 &transaction,
742 shared.key.as_ref(),
743 None,
744 batch,
745 &edit.author,
746 &edit.edit,
747 )?);
748 }
749 Ok((batch, transaction.commit().err()))
750 });
751 // The error, and the edit it refuses when it would fail again.
752 let written = match committed {
753 Ok((batch, None)) => Ok(batch),
754 Ok((batch, Some(error))) => {
755 if landed(&mut connection, ids.last().copied().unwrap_or_default())? {
756 Ok(batch)
757 } else {
758 Err((error.into(), None))
774 }759 }
775 }760 }
776 Err(error) => {761 Err(error) => {
762 let refused = (!transient(&error)).then_some(ids.len());
763 Err((error, refused))
764 }
765 };
766 drop(connection);
767 let batch = match written {
768 Ok(batch) => batch,
769 Err((error, refused)) => {
777 let message = error.to_string();770 let message = error.to_string();
778 for edit in accepted.drain(..) {771 let before = self.unwritten.len();
772 let mut at = 0;
773 for edit in self.unwritten.extract_if(.., |edit| {
774 at += 1;
775 edit.waiting.is_none() || refused == Some(at - 1)
776 }) {
779 (edit.reply)(Err(io::Error::other(message.clone()).into()));777 (edit.reply)(Err(io::Error::other(message.clone()).into()));
780 }778 }
779 if let Some(notice) = self
780 .unwritten
781 .first()
782 .and_then(|edit| edit.waiting.as_ref())
783 {
784 notice(&message);
785 }
786 return if self.unwritten.len() == before {
787 Ok(())
788 } else {
789 Err(error)
790 };
791 }
792 };
793 self.open = Some(batch);
794 #[cfg(target_arch = "wasm32")]
795 shared.receipt(
796 self.unwritten
797 .drain(..)
798 .zip(ids)
799 .map(|(edit, id)| {
800 let reply: Reply<()> =
801 Box::new(move |result| (edit.reply)(result.map(|()| id)));
802 reply
803 })
804 .collect(),
805 true,
806 );
807 #[cfg(not(target_arch = "wasm32"))]
808 {
809 for (edit, id) in self.unwritten.drain(..).zip(ids) {
810 (edit.reply)(Ok(id));
781 }811 }
812 crate::edited(&shared.worker);
782 }813 }
783 ok814 Ok(())
784 }815 }
785816
786 /// Seals the open batch unless a sealed batch still waits for publication.817 /// Seals the open batch unless a sealed batch still waits for publication.
...@@ -1052,3 +1083,37 @@ pub(crate) fn sealed(connection: &Connection) -> Result<Option<Sealed>> {...@@ -1052,3 +1083,37 @@ pub(crate) fn sealed(connection: &Connection) -> Result<Option<Sealed>> {
1052 })1083 })
1053 .transpose()1084 .transpose()
1054}1085}
1086
1087/// Whether a write that failed may succeed when tried again, once the disk has room or a lock
1088/// is let go; any other error would fail the same way forever.
1089fn transient(error: &crate::Error) -> bool {
1090 use rusqlite::ErrorCode::*;
1091 matches!(error, crate::Error::Database(rusqlite::Error::SqliteFailure(error, _))
1092 if matches!(error.code, SystemIoFailure | DiskFull | DatabaseBusy | DatabaseLocked | CannotOpen))
1093}
1094
1095/// Whether the commit whose last edit is `last` reached the cache after it failed, as a cold
1096/// open reads it, a commit landing whole: the warm connection may not know of frames its WAL
1097/// holds, and must never write again. When the cache cannot be reopened, an inert connection
1098/// stays, failing every statement.
1099fn landed(connection: &mut Connection, last: u64) -> Result<bool> {
1100 let path = connection.path().map(PathBuf::from);
1101 let inert = Connection::open_in_memory()?;
1102 inert.execute_batch("PRAGMA query_only=1")?;
1103 std::mem::replace(connection, inert)
1104 .close()
1105 .map_err(|(_, error)| error)?;
1106 // Reopening through the browser's file system is unverified: it stays closed.
1107 let path = path
1108 .filter(|_| cfg!(not(target_arch = "wasm32")))
1109 .ok_or_else(|| io::Error::other("The cache could not confirm a write"))?;
1110 let cold = crate::cache_connection(&path)?;
1111 crate::write_ahead(&cold)?;
1112 let found: bool = cold.query_row(
1113 "SELECT EXISTS(SELECT 1 FROM edits WHERE id=?1)",
1114 [crate::signed(last)?],
1115 |row| row.get(0),
1116 )?;
1117 *connection = cold;
1118 Ok(found)
1119}
crates/notebook/tests/commit_fault.rs created+177
...@@ -0,0 +1,177 @@
1//! Commits whose WAL write or sync fails, through a shim over the platform's VFS. Its own
2//! binary, as the shim becomes the process's default VFS.
3#![cfg(unix)]
4
5use notebook::{
6 Replica,
7 session::{Event, Section},
8};
9use onestore::op::{Edit, Op, PageOp};
10use std::{path::Path, sync::atomic::Ordering, sync::mpsc, time::Duration};
11
12#[path = "support/vfs.rs"]
13mod vfs;
14use vfs::{
15 ARMED, KEEP_WAL, OPEN, SYNC_AFTER, SYNC_BEFORE, WAL_WRITES_AFTER_FAULT, WRITE_COMMIT, install,
16};
17
18fn colour(space: onestore::ExGuid, colour: u32) -> Edit {
19 Edit {
20 at: 133_000_000_000_000_000 + u64::from(colour),
21 ops: vec![Op::Page {
22 space,
23 op: PageOp::Color(Some(colour)),
24 }],
25 }
26}
27
28/// What the section answered for the edit committed under `fault` and the one after it,
29/// whether a failed write was reported, and the queue as the section saw it.
30struct Outcome {
31 replies: [Result<u64, String>; 2],
32 failed: bool,
33 pending: Option<Vec<u64>>,
34}
35
36/// Seeds the WAL with one edit, so the armed commit's sync is the only one it makes, then
37/// commits under `fault` and once more after it; with `crash`, leaves the cache open.
38fn scenario(root: &Path, fault: u8, crash: bool) -> Outcome {
39 install();
40 let replica = Replica::open(root.join("replica.sqlite")).unwrap();
41 let section = Section::resume(root.join("Fault.one"), replica, true, || {}).unwrap();
42 let space = section.pages().unwrap()[0].0;
43 let apply = |value| {
44 let (sender, receiver) = mpsc::channel();
45 section
46 .apply_with("Fixture", colour(space, value), move |result| {
47 sender.send(result).unwrap();
48 })
49 .unwrap();
50 receiver
51 .recv_timeout(Duration::from_secs(10))
52 .unwrap()
53 .map_err(|error| error.to_string())
54 };
55 apply(1).unwrap();
56 WAL_WRITES_AFTER_FAULT.store(0, Ordering::SeqCst);
57 ARMED.store(fault, Ordering::SeqCst);
58 let first = apply(2);
59 assert_eq!(ARMED.load(Ordering::SeqCst), 0, "fault {fault} never fired");
60 assert_eq!(
61 WAL_WRITES_AFTER_FAULT.load(Ordering::SeqCst),
62 0,
63 "the warm connection wrote frames after its commit failed"
64 );
65 let second = apply(3);
66 let failed = section
67 .events()
68 .iter()
69 .any(|event| matches!(event, Event::Failed(_)));
70 let pending = section
71 .pending()
72 .ok()
73 .map(|pending| pending.iter().map(|edit| edit.id).collect());
74 if crash {
75 std::mem::forget(section);
76 } else {
77 section.close().unwrap();
78 }
79 Outcome {
80 replies: [first, second],
81 failed,
82 pending,
83 }
84}
85
86#[test]
87fn commit_fault_child() {
88 let (Some(root), Some(fault)) = (
89 std::env::var_os("COMMIT_FAULT_CHILD"),
90 std::env::var("COMMIT_FAULT").ok(),
91 ) else {
92 return;
93 };
94 let root = Path::new(&root);
95 let outcome = scenario(root, fault.parse().unwrap(), true);
96 let ids = |result: &Result<u64, String>| result.as_ref().map_or(0, |id| *id);
97 std::fs::write(
98 root.join("outcome"),
99 format!(
100 "{} {} {}",
101 ids(&outcome.replies[0]),
102 ids(&outcome.replies[1]),
103 outcome.failed
104 ),
105 )
106 .unwrap();
107 // Ends as a crash would: no SQLite destructor checkpoints or removes the WAL.
108 std::process::exit(0);
109}
110
111#[test]
112fn a_failed_commit_is_decided_by_a_cold_read_and_never_replayed_twice() {
113 install();
114 // (fault, whether the edit reached the cache, whether the cache stays closed)
115 for (fault, landed, closed) in [
116 (SYNC_BEFORE, false, false),
117 (SYNC_AFTER, false, false),
118 (WRITE_COMMIT, false, false),
119 (SYNC_AFTER | KEEP_WAL, true, false),
120 (SYNC_BEFORE | OPEN, false, true),
121 ] {
122 for crash in [false, true] {
123 let directory = tempfile::tempdir().unwrap();
124 let root = directory.path();
125 let source = onestore::create_section("Fault.one", "Original", "Fixture").unwrap();
126 std::fs::write(root.join("Fault.one"), &source).unwrap();
127 drop(Replica::create(root.join("replica.sqlite"), &source).unwrap());
128 let (first, second, failed) = if crash {
129 let output = std::process::Command::new(std::env::current_exe().unwrap())
130 .args(["--exact", "commit_fault_child", "--nocapture"])
131 .env("COMMIT_FAULT_CHILD", root)
132 .env("COMMIT_FAULT", fault.to_string())
133 .output()
134 .unwrap();
135 assert!(output.status.success(), "{output:?}");
136 let outcome = std::fs::read_to_string(root.join("outcome")).unwrap();
137 let mut fields = outcome.split(' ');
138 let mut id = || Some(fields.next()?.parse::<u64>().unwrap()).filter(|id| *id > 0);
139 let (first, second) = (id(), id());
140 (first, second, fields.next() == Some("true"))
141 } else {
142 let outcome = scenario(root, fault, false);
143 let [first, second] = outcome.replies;
144 if !closed {
145 let ids = [first.clone().unwrap(), second.clone().unwrap()];
146 assert_eq!(&outcome.pending.unwrap()[1..], ids);
147 } else {
148 assert!(outcome.pending.is_none());
149 }
150 (first.ok(), second.ok(), outcome.failed)
151 };
152 assert_eq!(failed, !landed, "fault {fault}");
153 let cold = Replica::open(root.join("replica.sqlite")).unwrap();
154 let colours: Vec<_> = cold
155 .pending()
156 .unwrap()
157 .iter()
158 .map(|edit| match &edit.edit.ops[..] {
159 [
160 Op::Page {
161 op: PageOp::Color(Some(colour)),
162 ..
163 },
164 ] => (edit.id, *colour),
165 ops => panic!("{ops:?}"),
166 })
167 .collect();
168 if closed {
169 assert_eq!((first, second), (None, None));
170 assert_eq!(colours.len(), 1, "fault {fault}");
171 } else {
172 let (first, second) = (first.unwrap(), second.unwrap());
173 assert_eq!(&colours[1..], [(first, 2), (second, 3)], "fault {fault}");
174 }
175 }
176 }
177}
crates/notebook/tests/support/vfs.rs created+147
...@@ -0,0 +1,147 @@
1//! A VFS over the platform's, made the process's default by `install`, that fails what a
2//! test arms: once, the WAL's sync or commit frame or the next database open; or every WAL
3//! write of the database `FAILING` names, for as long as it names it.
4#![allow(dead_code)]
5
6use rusqlite::ffi;
7use std::{
8 ffi::{CStr, c_char, c_int, c_void},
9 path::PathBuf,
10 sync::{
11 Mutex, Once, OnceLock,
12 atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering},
13 },
14};
15
16/// The WAL's sync at commit fails before it reaches the disk.
17pub const SYNC_BEFORE: u8 = 1;
18/// The WAL's sync at commit reaches the disk and still reports failure.
19pub const SYNC_AFTER: u8 = 2;
20/// The commit frame's header is never written.
21pub const WRITE_COMMIT: u8 = 4;
22/// The next open of a database fails, as the cold reopen after a failed commit.
23pub const OPEN: u8 = 8;
24/// After a fault, the next sync fails, as the checkpoint closing the warm connection makes,
25/// which keeps the WAL.
26pub const KEEP_WAL: u8 = 16;
27
28pub static ARMED: AtomicU8 = AtomicU8::new(0);
29/// Between a fault and the next database open, as the replica reopens the cache cold.
30static FIRED: AtomicBool = AtomicBool::new(false);
31pub static WAL_WRITES_AFTER_FAULT: AtomicUsize = AtomicUsize::new(0);
32static REAL: OnceLock<usize> = OnceLock::new();
33/// The platform's WAL file methods, the shim's copy of them, and the copy failing every write.
34static WAL: OnceLock<(usize, usize, usize)> = OnceLock::new();
35/// While set, every write to this database's WAL fails, for WALs opened while it was set.
36pub static FAILING: Mutex<Option<PathBuf>> = Mutex::new(None);
37
38fn fire(fault: u8) -> bool {
39 let fired = ARMED.fetch_and(!fault, Ordering::SeqCst) & fault != 0;
40 if fired {
41 FIRED.store(true, Ordering::SeqCst);
42 }
43 fired
44}
45
46fn real_wal() -> &'static ffi::sqlite3_io_methods {
47 unsafe { &*(WAL.get().unwrap().0 as *const ffi::sqlite3_io_methods) }
48}
49
50unsafe extern "C" fn open(
51 _: *mut ffi::sqlite3_vfs,
52 name: *const c_char,
53 file: *mut ffi::sqlite3_file,
54 flags: c_int,
55 out: *mut c_int,
56) -> c_int {
57 if flags & ffi::SQLITE_OPEN_MAIN_DB != 0 {
58 FIRED.store(false, Ordering::SeqCst);
59 if fire(OPEN) {
60 FIRED.store(false, Ordering::SeqCst);
61 unsafe { (*file).pMethods = std::ptr::null() };
62 return ffi::SQLITE_CANTOPEN;
63 }
64 }
65 let real = *REAL.get().unwrap() as *mut ffi::sqlite3_vfs;
66 let opened = unsafe { (*real).xOpen.unwrap()(real, name, file, flags, out) };
67 if opened == ffi::SQLITE_OK && flags & ffi::SQLITE_OPEN_WAL != 0 {
68 let methods = unsafe { (*file).pMethods };
69 let &(platform, shim, failing) = WAL.get_or_init(|| {
70 let mut shim = unsafe { *methods };
71 shim.xWrite = Some(write);
72 shim.xSync = Some(sync);
73 let mut failing = shim;
74 failing.xWrite = Some(fail);
75 let leak = |methods| Box::leak(Box::new(methods)) as *const _ as usize;
76 (methods as usize, leak(shim), leak(failing))
77 });
78 assert_eq!(platform, methods as usize);
79 let name = unsafe { CStr::from_ptr(name) }.to_string_lossy();
80 let fails = FAILING
81 .lock()
82 .unwrap()
83 .as_ref()
84 .is_some_and(|database| name.starts_with(&*database.to_string_lossy()));
85 let methods = if fails { failing } else { shim };
86 unsafe { (*file).pMethods = methods as *const ffi::sqlite3_io_methods };
87 }
88 opened
89}
90
91unsafe extern "C" fn write(
92 file: *mut ffi::sqlite3_file,
93 data: *const c_void,
94 amount: c_int,
95 offset: ffi::sqlite3_int64,
96) -> c_int {
97 // A frame header names the database size only in the frame that commits.
98 if amount == 24
99 && unsafe { std::slice::from_raw_parts(data.cast::<u8>().add(4), 4) } != [0; 4]
100 && fire(WRITE_COMMIT)
101 {
102 return ffi::SQLITE_IOERR_WRITE;
103 }
104 if FIRED.load(Ordering::SeqCst) {
105 WAL_WRITES_AFTER_FAULT.fetch_add(1, Ordering::SeqCst);
106 }
107 unsafe { real_wal().xWrite.unwrap()(file, data, amount, offset) }
108}
109
110unsafe extern "C" fn fail(
111 file: *mut ffi::sqlite3_file,
112 data: *const c_void,
113 amount: c_int,
114 offset: ffi::sqlite3_int64,
115) -> c_int {
116 if FAILING.lock().unwrap().is_some() {
117 return ffi::SQLITE_IOERR_WRITE;
118 }
119 unsafe { write(file, data, amount, offset) }
120}
121
122unsafe extern "C" fn sync(file: *mut ffi::sqlite3_file, flags: c_int) -> c_int {
123 if fire(SYNC_BEFORE) || FIRED.load(Ordering::SeqCst) && fire(KEEP_WAL) {
124 return ffi::SQLITE_IOERR_FSYNC;
125 }
126 let synced = unsafe { real_wal().xSync.unwrap()(file, flags) };
127 if synced == ffi::SQLITE_OK && fire(SYNC_AFTER) {
128 return ffi::SQLITE_IOERR_FSYNC;
129 }
130 synced
131}
132
133pub fn install() {
134 static INSTALL: Once = Once::new();
135 INSTALL.call_once(|| unsafe {
136 let real = ffi::sqlite3_vfs_find(std::ptr::null());
137 REAL.set(real as usize).unwrap();
138 let mut shim = *real;
139 shim.zName = c"commit-fault".as_ptr();
140 shim.pNext = std::ptr::null_mut();
141 shim.xOpen = Some(open);
142 assert_eq!(
143 ffi::sqlite3_vfs_register(Box::leak(Box::new(shim)), 1),
144 ffi::SQLITE_OK
145 );
146 });
147}
crates/snowbound/src/main.rs+1-1
...@@ -4421,7 +4421,7 @@ impl State {...@@ -4421,7 +4421,7 @@ impl State {
4421 eprintln!("Saving failed: {error}");4421 eprintln!("Saving failed: {error}");
4422 }4422 }
4423 }4423 }
4424 Event::Failed(error) => eprintln!("Synchronization stopped: {error}"),4424 Event::Failed(error) => eprintln!("Not saving: {error}"),
4425 // Its guests hear of it sooner than the folder's watch would tell them.4425 // Its guests hear of it sooner than the folder's watch would tell them.
4426 #[cfg(feature = "live")]4426 #[cfg(feature = "live")]
4427 Event::Attempt {4427 Event::Attempt {