From fe5ea9c240ad4fc12520c83370d4938f97ff87c5 Mon Sep 17 00:00:00 2001 From: clover caruso Date: Tue, 6 Oct 2026 00:26:55 -0700 Subject: [PATCH] fix(notebook): durably acknowledge offline browser edits Await ordered OPFS receipts without suspending logical section reads. Migrate browser edit, history, recycle and password callers to async continuations while keeping native synchronous completion. Retain cache ownership through pending receipts and release it after SQLite closes. Await destination durability before destructive page operations, including recycle-bin writes. Keep shown and kept models available on storage failure. Native notebook/snowbound all-feature cargo check passed; Node storage bridge regression passed (1 test). Exact native tests and wasm gate are still running or pending; root owns real renderer-crash and caller-path verification. Depends on #97 recoverable storage retries. Fixes #98 Assisted-by: gpt-6.1-sol --- crates/notebook/src/background.rs | 2 +- crates/notebook/src/lib.rs | 50 ++++-- crates/notebook/src/session.rs | 165 +++++++++++++----- crates/notebook/src/task.rs | 63 +++++-- crates/notebook/src/worker.rs | 16 ++ crates/notebook/src/working.rs | 151 +++++++++++++++-- crates/notebook/tests/cache.rs | 35 ++++ crates/snowbound/src/history.rs | 125 ++++++++------ crates/snowbound/src/library.rs | 48 +++++- crates/snowbound/src/main.rs | 84 ++++++++-- crates/snowbound/src/manage.rs | 4 +- crates/snowbound/src/protection.rs | 259 +++++++++++++++++++++-------- crates/snowbound/src/recycle.rs | 63 ++++--- crates/snowbound/src/undo.rs | 84 ++++++---- crates/snowbound/src/web.rs | 9 +- crates/snowbound/web/glue.js | 2 +- tools/web/test_storage_barrier.mjs | 59 +++++++ 17 files changed, 936 insertions(+), 283 deletions(-) create mode 100644 tools/web/test_storage_barrier.mjs diff --git a/crates/notebook/src/background.rs b/crates/notebook/src/background.rs index 10bff93ca3c80f474c459d65394677662a07816c..105207df8b680e21990c68644c35750b6431ebef 100644 --- a/crates/notebook/src/background.rs +++ b/crates/notebook/src/background.rs @@ -433,7 +433,7 @@ impl Background { } } let (shared, path) = (Arc::downgrade(&self.0), path.to_owned()); - if let Ok(mut released) = replica.released.0.lock() { + if let Ok(mut released) = replica.section.released.0.lock() { *released = Some(Box::new(move || { if let Some(shared) = shared.upgrade() { shared.released(&path); diff --git a/crates/notebook/src/lib.rs b/crates/notebook/src/lib.rs index b8187f649c656f1be58983c7c5a6f044cfbb9afc..3fdc43864fe577bb6bf791c51731c4031d6263ab 100644 --- a/crates/notebook/src/lib.rs +++ b/crates/notebook/src/lib.rs @@ -112,8 +112,6 @@ pub struct Replica { synchronization: Mutex<()>, /// The section's root object space, which names the document. root: ExGuid, - /// Last, so that it runs once the fields above have closed the cache. - released: Released, } /// Runs once the replica is released and its cache closed, as `Background::hold` asks. @@ -254,7 +252,6 @@ impl Replica { section, synchronization: Mutex::new(()), root, - released: Released::default(), }) } @@ -281,18 +278,39 @@ impl Replica { .map_err(|_| io::Error::other("The section thread stopped"))? } + async fn ask_async( + &self, + request: impl FnOnce(working::Reply) -> working::Request + Send, + ) -> Result { + let (sender, receiver) = task::response(); + self.send(request(Box::new(move |result| { + let _ = sender.send(result); + })))?; + #[cfg(target_arch = "wasm32")] + let answer = task::receive(&receiver).await; + #[cfg(not(target_arch = "wasm32"))] + let answer = receiver.recv(); + answer.map_err(|_| io::Error::other("The section thread stopped"))? + } + /// Applies an edit and queues it durably for publication, returning its id once /// written. A refused edit returns `Rejected` and leaves every page as it was. + /// Browser callers use `apply_async`; this returns `WouldBlock` without submitting. pub fn apply(&self, author: &str, edit: Edit) -> Result { - self.ask(|reply| working::Request::Apply { + task::blocking(self.apply_async(author, edit))? + } + + /// 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, }) + .await } - /// `apply` without waiting: `reply` runs on the section thread once the edit is - /// durable or refused. + /// `apply` without waiting: `reply` runs once the edit is durable or refused. pub(crate) fn submit( &self, author: &str, @@ -338,14 +356,26 @@ impl Replica { /// The section image the queued edits leave, the unsealed ones sealed as one more /// revision whose identities differ per call: O(section). pub fn snapshot(&self) -> Result> { - self.written()?; + self.ask(|reply| working::Request::Flush { + reply, + durable: false, + })?; working::image(&*self.lock()?, self.section.key.as_ref()) } - /// Waits until the edits applied before it are written to the queue, as reads answer - /// before an open burst of edits is. + /// Waits until the edits applied before it are durable in the local queue. + /// Browser callers use `written_async`; this returns `WouldBlock` without submitting. pub fn written(&self) -> Result<()> { - self.ask(|reply| working::Request::Flush { reply }) + task::blocking(self.written_async())? + } + + /// Waits for durable local storage of the edits submitted before this request. + pub async fn written_async(&self) -> Result<()> { + self.ask_async(|reply| working::Request::Flush { + reply, + durable: true, + }) + .await } /// The section file's identity, which internal links name as `section-id`. diff --git a/crates/notebook/src/session.rs b/crates/notebook/src/session.rs index 107153511fef8e69ffd93154e814538dcec39f8c..5d0600ca57ccc2c1dadf774ff4fe43b6c8a4371d 100644 --- a/crates/notebook/src/session.rs +++ b/crates/notebook/src/session.rs @@ -1347,6 +1347,10 @@ impl Notebook { /// queued again under the key: each page they changed comes back as a copy, as a page /// another client removed does, the section's pages having taken new identities. pub fn unlock(&self, path: &str, password: &str) -> Result { + crate::task::blocking(self.unlock_async(path, password))? + } + + pub async fn unlock_async(&self, path: &str, password: &str) -> Result { let section = self.section_path(path)?; let image = self.storage.read(§ion.path)?; let key = protected::Key::open(&image, password)?; @@ -1365,7 +1369,7 @@ impl Notebook { continue; }; for (author, edit) in old.copies()? { - replica.apply(&author, edit)?; + replica.apply_async(&author, edit).await?; } drop(old); remove_replica(&path)?; @@ -2242,31 +2246,60 @@ impl Section { /// fresh identities, queued like the user's own edits. Returns the new page's space. /// Content outside the model refuses to copy. pub fn import_page(&self, page: &Page, author: &str) -> Result { - let creation = PageCreation::new(None, Some(&page.title), author)?; - let space = creation.space(); - self.replica.apply( - author, - Edit { - at: crate::now(), - ops: vec![Op::Section(SectionOp::Import { - creation, - page: page.copy()?, - })], - }, - )?; - Ok(space) + crate::task::blocking(self.import_page_async(page, author))? + } + + pub fn import_page_async( + &self, + page: &Page, + author: &str, + ) -> impl Future> + Send + 'static + use<> { + let replica = Arc::clone(&self.replica); + let author = author.to_owned(); + let creation = PageCreation::new(None, Some(&page.title), &author); + let page = page.copy(); + async move { + let creation = creation?; + let space = creation.space(); + replica + .apply_async( + &author, + Edit { + at: crate::now(), + ops: vec![Op::Section(SectionOp::Import { + creation, + page: page?, + })], + }, + ) + .await?; + Ok(space) + } } /// Removes pages or conflict pages permanently, queued like the user's own edits (a /// move across sections is `import_page` there, then this here). pub fn delete_pages(&self, pages: &[ExGuid]) -> Result { - self.replica.apply( - "", - Edit { - at: crate::now(), - ops: vec![Op::Section(SectionOp::Delete(pages.to_vec()))], - }, - ) + crate::task::blocking(self.delete_pages_async(pages))? + } + + pub fn delete_pages_async( + &self, + pages: &[ExGuid], + ) -> impl Future> + Send + 'static + use<> { + let replica = Arc::clone(&self.replica); + let pages = pages.to_vec(); + async move { + replica + .apply_async( + "", + Edit { + at: crate::now(), + ops: vec![Op::Section(SectionOp::Delete(pages))], + }, + ) + .await + } } pub fn status(&self, id: u64) -> Result> { @@ -2302,32 +2335,61 @@ impl Section { /// Makes a page's version its current state, the page as it stood becoming the newest /// version, queued like the user's own edits. pub fn restore_version(&self, space: ExGuid, version: ExGuid, author: &str) -> Result { - self.replica.apply( - author, - Edit { - at: crate::now(), - ops: vec![Op::Section(SectionOp::restore(space, version)?)], - }, - ) + crate::task::blocking(self.restore_version_async(space, version, author))? + } + + pub fn restore_version_async( + &self, + space: ExGuid, + version: ExGuid, + author: &str, + ) -> impl Future> + Send + 'static + use<> { + let replica = Arc::clone(&self.replica); + let author = author.to_owned(); + let op = SectionOp::restore(space, version); + async move { + replica + .apply_async( + &author, + Edit { + at: crate::now(), + ops: vec![Op::Section(op?)], + }, + ) + .await + } } /// Deletes versions of pages, queued like the user's own edits. pub fn delete_versions(&self, versions: &[(ExGuid, Vec)]) -> Result { - self.replica.apply( - "", - Edit { - at: crate::now(), - ops: versions - .iter() - .map(|(page, versions)| { - Op::Section(SectionOp::DeleteVersions { - page: *page, - versions: versions.clone(), - }) - }) - .collect(), - }, - ) + crate::task::blocking(self.delete_versions_async(versions))? + } + + pub fn delete_versions_async( + &self, + versions: &[(ExGuid, Vec)], + ) -> impl Future> + Send + 'static + use<> { + let replica = Arc::clone(&self.replica); + let ops = versions + .iter() + .map(|(page, versions)| { + Op::Section(SectionOp::DeleteVersions { + page: *page, + versions: versions.clone(), + }) + }) + .collect(); + async move { + replica + .apply_async( + "", + Edit { + at: crate::now(), + ops, + }, + ) + .await + } } /// Retires an uncertain attempt after review, exporting the queue to `archive` first: @@ -2409,6 +2471,23 @@ impl Section { None => Ok(()), } } + + /// Stops synchronization without consuming the section's recoverable model. + pub fn stop_async(&mut self) -> impl Future> + Send + 'static + use<> { + let worker = self.worker.take(); + async move { + match worker { + Some(worker) => worker.stop_async().await, + None => Ok(()), + } + } + } + + /// Makes the submitted edits locally durable and awaits synchronization before closing. + pub async fn close_async(mut self) -> Result<()> { + self.replica.written_async().await?; + self.stop_async().await + } } /// The section file itself as the publication target, under OneNote-compatible exclusion. diff --git a/crates/notebook/src/task.rs b/crates/notebook/src/task.rs index 77f725750996aed7dfd7b775b06e42a54fb25894..df2ac9396dd820c2fa73cfe9e332e139c2090771 100644 --- a/crates/notebook/src/task.rs +++ b/crates/notebook/src/task.rs @@ -10,14 +10,25 @@ pub(crate) use std::{ thread::JoinHandle, }; -#[cfg(all(feature = "live", not(target_arch = "wasm32")))] +#[cfg(not(target_arch = "wasm32"))] pub(crate) type Answer = std::sync::mpsc::Sender; -#[cfg(all(feature = "live", not(target_arch = "wasm32")))] +#[cfg(not(target_arch = "wasm32"))] pub(crate) fn response() -> (Answer, std::sync::mpsc::Receiver) { std::sync::mpsc::channel() } +/// Runs native blocking work; browser callers are refused before the future is polled. +pub(crate) fn blocking(work: impl Future) -> io::Result { + #[cfg(not(target_arch = "wasm32"))] + return Ok(complete(work)); + #[cfg(target_arch = "wasm32")] + { + drop(work); + Err(io::ErrorKind::WouldBlock.into()) + } +} + #[cfg(all(feature = "live", not(target_arch = "wasm32")))] pub(crate) async fn answered( receiver: &std::sync::mpsc::Receiver, @@ -134,27 +145,43 @@ mod web { } } + #[cfg(feature = "live")] pub(crate) async fn answered( receiver: &Answered, timeout: Duration, ) -> Result { let deadline = web_time::Instant::now() + timeout; let mut armed = false; + let mut answer = std::pin::pin!(receive(receiver)); + std::future::poll_fn(|context| { + if let Poll::Ready(answer) = answer.as_mut().poll(context) { + return Poll::Ready( + answer.map_err(|_| std::sync::mpsc::RecvTimeoutError::Disconnected), + ); + } + if web_time::Instant::now() >= deadline { + return Poll::Ready(Err(std::sync::mpsc::RecvTimeoutError::Timeout)); + } + if !std::mem::replace(&mut armed, true) { + after(timeout, context.waker().clone()); + } + Poll::Pending + }) + .await + } + + pub(crate) async fn receive( + receiver: &Answered, + ) -> Result { std::future::poll_fn(|context| { let mut response = receiver.0.lock().unwrap(); if let Some(value) = response.value.take() { return Poll::Ready(Ok(value)); } if response.closed { - return Poll::Ready(Err(std::sync::mpsc::RecvTimeoutError::Disconnected)); - } - if web_time::Instant::now() >= deadline { - return Poll::Ready(Err(std::sync::mpsc::RecvTimeoutError::Timeout)); + return Poll::Ready(Err(std::sync::mpsc::RecvError)); } response.waiting = Some(context.waker().clone()); - if !std::mem::replace(&mut armed, true) { - after(timeout, context.waker().clone()); - } Poll::Pending }) .await @@ -229,13 +256,17 @@ mod web { .await } - /// A task the browser's event loop runs. It cannot be waited for, and ends on its own. - pub(crate) struct JoinHandle(Arc>>); + /// A task the browser's event loop runs, with a wakeable completion receipt. + pub(crate) struct JoinHandle(Answered); impl JoinHandle { /// What the task ended with, if it has ended. pub(crate) fn finished(self) -> Option { - self.0.lock().ok()?.take() + self.0.0.lock().ok()?.value.take() + } + + pub(crate) async fn join(self) -> Result { + receive(&self.0).await } } @@ -243,14 +274,10 @@ mod web { _: &str, start: impl FnOnce() -> F, ) -> io::Result> { - let output = Arc::new(Mutex::new(None)); - let ended = Arc::clone(&output); + let (ended, output) = response(); let work = start(); wasm_bindgen_futures::spawn_local(async move { - let result = work.await; - if let Ok(mut ended) = ended.lock() { - *ended = Some(result); - } + let _ = ended.send(work.await); }); Ok(JoinHandle(output)) } diff --git a/crates/notebook/src/worker.rs b/crates/notebook/src/worker.rs index 6d3028472385017628f106a2d8ec58c542b05798..df6ae0fd73a40e4c0cb805890ee505de559bb7b6 100644 --- a/crates/notebook/src/worker.rs +++ b/crates/notebook/src/worker.rs @@ -130,6 +130,22 @@ impl SyncWorker { .join() .map_err(|_| io::Error::other("Synchronization worker panicked"))? } + + pub(crate) async fn stop_async(self) -> Result<()> { + #[cfg(not(target_arch = "wasm32"))] + return self.stop(); + #[cfg(target_arch = "wasm32")] + { + let mut worker = self; + worker.signal.stopped.store(true, Ordering::Release); + worker.signal.wake(); + let thread = worker.thread.take().expect("Worker owns its thread"); + thread + .join() + .await + .map_err(|_| io::Error::other("Synchronization worker panicked"))? + } + } } impl Drop for SyncWorker { diff --git a/crates/notebook/src/working.rs b/crates/notebook/src/working.rs index df13e01b9cfbc3bff6ff2733b4be1e161ee8715d..0c76eabb12fdd3825cf7b331e95abde8b091cab3 100644 --- a/crates/notebook/src/working.rs +++ b/crates/notebook/src/working.rs @@ -44,9 +44,10 @@ pub(crate) enum Request { version: ExGuid, reply: Reply, }, - /// Answers once the edits before it are written. + /// Answers once the edits before it are written, optionally awaiting host durability. Flush { reply: Reply<()>, + durable: bool, }, /// Seals the open batch, when no sealed batch waits for publication, recording the /// attempt to publish it. @@ -90,7 +91,7 @@ impl Request { Self::Conflicts { reply } => reply(Err(error())), Self::Versions { reply } => reply(Err(error())), Self::Version { reply, .. } => reply(Err(error())), - Self::Flush { reply } => reply(Err(error())), + Self::Flush { reply, .. } => reply(Err(error())), Self::Seal { reply } => reply(Err(error())), Self::Rebase { reply, .. } => reply(Err(error())), Self::Reopen { reply } => reply(Err(error())), @@ -109,10 +110,16 @@ pub(crate) struct Thread { pub(crate) worker: Mutex>, /// Taken when the replica drops, which ends every section thread. sender: Mutex>>, + #[cfg(target_arch = "wasm32")] + durable: Mutex>>, + #[cfg(target_arch = "wasm32")] + durable_wake: crate::task::Sender<()>, #[cfg(not(target_arch = "wasm32"))] threads: Mutex>>, /// A password-protected section's key, under which its queue is sealed too. pub(crate) key: Option, + /// Last, so the cache closes before background synchronization takes it over. + pub(crate) released: crate::Released, } impl Thread { @@ -127,13 +134,19 @@ impl Thread { Ok(()) } - /// Ends the section threads and waits for them, so the cache is released. + /// Ends request serving; pending browser receipts retain the cache until they settle. pub(crate) fn stop(&self) { if let Ok(mut sender) = self.sender.lock() { sender.take(); } #[cfg(target_arch = "wasm32")] - serving::serve(self); + { + serving::serve(self); + if let Ok(mut durable) = self.durable.lock() { + durable.take(); + } + let _ = self.durable_wake.try_send(()); + } #[cfg(not(target_arch = "wasm32"))] while let Some(thread) = self .threads @@ -169,6 +182,77 @@ impl Thread { serving::start(self, Box::pin(run(Arc::clone(self)))); Ok(()) } + + #[cfg(target_arch = "wasm32")] + fn receipt(self: &Arc, replies: Vec>, edited: bool) { + let receipt = Receipt { + shared: Arc::clone(self), + replies, + edited, + }; + let sent = match self.durable.lock() { + Ok(sender) => match sender.as_ref() { + Some(sender) => sender.send(receipt).map_err(|error| error.0), + None => Err(receipt), + }, + Err(_) => Err(receipt), + }; + if let Err(receipt) = sent { + for reply in receipt.replies { + reply(Err(io::Error::other("The section thread stopped").into())); + } + } + let _ = self.durable_wake.try_send(()); + } +} + +#[cfg(target_arch = "wasm32")] +struct Receipt { + shared: Arc, + replies: Vec>, + edited: bool, +} + +#[cfg(target_arch = "wasm32")] +async fn receipts(requests: mpsc::Receiver, wake: crate::task::Receiver<()>) { + loop { + let Receipt { + shared, + replies, + edited, + } = match requests.try_recv() { + Ok(receipt) => receipt, + Err(mpsc::TryRecvError::Empty) => { + crate::task::wait(&wake, None).await; + continue; + } + Err(mpsc::TryRecvError::Disconnected) => return, + }; + let result = loop { + match crate::fs::durable().await { + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + crate::task::wait(&wake, Some(std::time::Duration::from_secs(1))).await; + } + result => break result, + } + }; + match result { + Ok(()) => { + for reply in replies { + reply(Ok(())); + } + if edited { + crate::edited(&shared.worker); + } + } + Err(error) => { + let message = error.to_string(); + for reply in replies { + reply(Err(io::Error::new(error.kind(), message.clone()).into())); + } + } + } + } } /// The browser's section threads: each replica's `run`, polled whenever a request is sent @@ -225,14 +309,25 @@ async fn receive(requests: &mpsc::Receiver) -> Option { /// Starts the section thread once the queue opens. pub(crate) fn spawn(connection: Connection, key: Option) -> Result<(Arc, ExGuid)> { let (sender, requests) = mpsc::channel(); + #[cfg(target_arch = "wasm32")] + let (durable, written) = mpsc::channel(); + #[cfg(target_arch = "wasm32")] + let (durable_wake, woken) = crate::task::channel(); let shared = Arc::new(Thread { connection: Mutex::new(connection), worker: Mutex::new(Weak::new()), sender: Mutex::new(Some(sender)), + #[cfg(target_arch = "wasm32")] + durable: Mutex::new(Some(durable)), + #[cfg(target_arch = "wasm32")] + durable_wake, #[cfg(not(target_arch = "wasm32"))] threads: Mutex::new(Vec::new()), key, + released: crate::Released::default(), }); + #[cfg(target_arch = "wasm32")] + crate::task::spawn("onestore-durable", move || receipts(written, woken))?; let (ready, opened) = mpsc::sync_channel(1); shared.start(move |shared| run(shared, requests, VecDeque::new(), Some(ready)))?; match opened.recv() { @@ -486,13 +581,25 @@ impl<'a> Working<'a> { later.push_back(request); false } - Request::Flush { reply } => { + Request::Flush { reply, durable } => { let written = self.flush(shared, &mut accepted); - reply(if written { - Ok(()) + 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(())); + } } else { - Err(io::Error::other("The queue could not be written").into()) - }); + reply(Err( + io::Error::other("The queue could not be written").into() + )); + } !written } Request::Seal { reply } => { @@ -586,7 +693,7 @@ impl<'a> Working<'a> { edit: Edit, reply: Reply, accepted: &mut Vec, - shared: &Thread, + shared: &Arc, ) -> std::result::Result<(), bool> { match self.section.apply(&author, &edit) { Ok(()) => { @@ -612,7 +719,7 @@ impl<'a> Working<'a> { /// 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: &Thread, accepted: &mut Vec) -> bool { + fn flush(&mut self, shared: &Arc, accepted: &mut Vec) -> bool { if accepted.is_empty() { return true; } @@ -645,10 +752,26 @@ impl<'a> Working<'a> { let ok = written.is_ok(); match written { Ok(ids) => { - for (edit, id) in accepted.drain(..).zip(ids) { - (edit.reply)(Ok(id)); + #[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)); + } + crate::edited(&shared.worker); } - crate::edited(&shared.worker); } Err(error) => { let message = error.to_string(); diff --git a/crates/notebook/tests/cache.rs b/crates/notebook/tests/cache.rs index f547954ad14140124ec2f7e612965d2ef622d6b7..400fe607af1ff6599aa51b6896ec83bb98264e25 100644 --- a/crates/notebook/tests/cache.rs +++ b/crates/notebook/tests/cache.rs @@ -132,6 +132,41 @@ fn cache_reopen_preserves_the_base_and_queued_edits() { assert_eq!(replica.pending().unwrap().len(), 2); } +#[test] +fn async_edit_receipts_reopen_with_the_same_id_and_text() { + use std::{ + future::Future, + task::{Context, Poll, Waker}, + }; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("async.sqlite"); + let source = onestore::create_section("Async.one", "Receipt", "Fixture").unwrap(); + let (space, text, _) = target(&source); + let replica = Replica::create(&path, &source).unwrap(); + let id = { + let mut receipt = + std::pin::pin!(replica.apply_async("Author", typed(space, text, "Durable "))); + match receipt + .as_mut() + .poll(&mut Context::from_waker(Waker::noop())) + { + Poll::Ready(result) => result.unwrap(), + Poll::Pending => panic!("Native durability must finish without a pending wait"), + } + }; + { + let mut flush = std::pin::pin!(replica.written_async()); + assert!(matches!( + flush.as_mut().poll(&mut Context::from_waker(Waker::noop())), + Poll::Ready(Ok(())) + )); + } + drop(replica); + let reopened = Replica::open(&path).unwrap(); + assert_eq!(reopened.pending().unwrap()[0].id, id); + assert_eq!(target(&reopened.snapshot().unwrap()).2, "Durable Receipt"); +} + #[test] fn refused_edits_and_failed_writes_leave_the_queue_as_it_was() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/snowbound/src/history.rs b/crates/snowbound/src/history.rs index 2380a6fa361ea257c644316c9e50f4d0fac2798f..da1a3e1f858a1284361493a4098f14d3bbfcedf1 100644 --- a/crates/snowbound/src/history.rs +++ b/crates/snowbound/src/history.rs @@ -169,13 +169,17 @@ impl State { version: ExGuid, ) -> Result<(), Box> { let session = self.session.as_mut().ok_or("No section is open")?; - session + let operation = session .section - .restore_version(page, version, &self.author)?; - session.refresh_history()?; - self.edited(vec![page]); - self.commands.push(Command::OpenPage(page)); - Ok(()) + .restore_version_async(page, version, &self.author); + self.section_edit(operation, move |state, _| { + if let Some(session) = &mut state.session { + session.refresh_history()?; + } + state.edited(vec![page]); + state.commands.push(Command::OpenPage(page)); + Ok(()) + }) } pub(crate) fn delete_version( @@ -184,10 +188,16 @@ impl State { version: ExGuid, ) -> Result<(), Box> { let session = self.session.as_mut().ok_or("No section is open")?; - session.section.delete_versions(&[(page, vec![version])])?; - session.refresh_history()?; - self.commands.push(Command::OpenPage(page)); - Ok(()) + let operation = session + .section + .delete_versions_async(&[(page, vec![version])]); + self.section_edit(operation, move |state, _| { + if let Some(session) = &mut state.session { + session.refresh_history()?; + } + state.commands.push(Command::OpenPage(page)); + Ok(()) + }) } /// Delete All Versions in the section, its group or the notebook, once confirmed. @@ -232,55 +242,70 @@ impl State { }) else { return Ok(()); }; - clear(&session.section)?; - session.refresh_history()?; - if session.version.is_some() { - self.commands.push(Command::OpenPage(session.space)); - } - if scope == Scope::Section { - return Ok(()); - } - // The other sections, each opened, cleared and closed in the background. - let folder = match scope { - Scope::Group => folder, - _ => String::new(), - }; + let operation = clear(&session.section); let library = Arc::clone(&session.library); let proxy = self.proxy.clone(); - crate::spawn(move || { - for path in sections(&library, &folder) { - if path == open { - continue; - } - let cleared = library - .open(&path, notify(proxy.clone())) - .and_then(|section| { - clear(§ion)?; - Ok(section.close()?) - }); - if let Err(error) = cleared { - eprintln!("Deleting the versions in {path} failed: {error}"); + let operation = async move { + operation.await?; + if scope != Scope::Section { + let folder = if scope == Scope::Group { + folder + } else { + String::new() + }; + crate::spawn_async(move || async move { + for path in sections(&library, &folder) { + if path == open { + continue; + } + let cleared: Result<(), Box> = async { + let section = library.open(&path, notify(proxy.clone()))?; + clear(§ion).await?; + Ok(section.close_async().await?) + } + .await; + if let Err(error) = cleared { + eprintln!("Deleting the versions in {path} failed: {error}"); + } + } + }); + } + Ok(()) + }; + self.section_edit(operation, move |state, ()| { + if let Some(session) = &mut state.session { + session.refresh_history()?; + if session.version.is_some() { + state.commands.push(Command::OpenPage(session.space)); } } - }); - Ok(()) + Ok(()) + }) } } /// Deletes every page version `section` holds. -fn clear(section: ¬ebook::session::Section) -> Result<(), notebook::Error> { - let all: Vec<(ExGuid, Vec)> = (section.versions()?.into_iter()) - .map(|(page, versions)| { - ( - page, - versions.iter().map(|version| version.context).collect(), - ) - }) - .collect(); - if !all.is_empty() { - section.delete_versions(&all)?; +fn clear( + section: ¬ebook::session::Section, +) -> impl std::future::Future> + Send + 'static + use<> { + let operation = section.versions().map(|versions| { + let all: Vec<(ExGuid, Vec)> = versions + .into_iter() + .map(|(page, versions)| { + ( + page, + versions.iter().map(|version| version.context).collect(), + ) + }) + .collect(); + (!all.is_empty()).then(|| section.delete_versions_async(&all)) + }); + async move { + if let Some(operation) = operation? { + operation.await?; + } + Ok(()) } - Ok(()) } /// The readable sections at catalog path `folder` and in the groups inside it, leaving out diff --git a/crates/snowbound/src/library.rs b/crates/snowbound/src/library.rs index 944eaa61f25b7112960b6dbf939fd06ad71e1320..5be7913fc05dd26487eaa99071355bf2791a7965 100644 --- a/crates/snowbound/src/library.rs +++ b/crates/snowbound/src/library.rs @@ -974,6 +974,41 @@ impl Library { } } + pub async fn close_kept_async(&self) -> Result<(), Box> { + #[cfg(not(target_arch = "wasm32"))] + { + self.close_kept(); + Ok(()) + } + #[cfg(target_arch = "wasm32")] + { + let replicas: Vec<_> = { + let mut kept = self + .kept + .lock() + .map_err(|_| io::Error::from(io::ErrorKind::Other))?; + kept.values_mut() + .map(|section| Arc::clone(section.replica())) + .collect() + }; + for replica in replicas { + replica.written_async().await?; + } + let stopping: Vec<_> = { + let mut kept = self + .kept + .lock() + .map_err(|_| io::Error::from(io::ErrorKind::Other))?; + kept.values_mut().map(Section::stop_async).collect() + }; + for stopping in stopping { + stopping.await?; + } + self.close_kept(); + Ok(()) + } + } + /// Where `file`, a section file of this notebook, sits on this computer: under the /// mount it was opened from when the embedded client holds it share-relative. pub fn local(&self, file: &Path) -> Option { @@ -1186,10 +1221,21 @@ impl Library { /// Unlocks the section at catalog `path` with `password`, as OneNote's Protected Section /// dialog does; a wrong password is `PasswordMismatch`. pub fn unlock(&self, path: &str, password: &str) -> Result<(), notebook::Error> { + let operation = self.unlock_async(path, password); + #[cfg(not(target_arch = "wasm32"))] + return pollster::block_on(operation); + #[cfg(target_arch = "wasm32")] + { + drop(operation); + Err(io::ErrorKind::WouldBlock.into()) + } + } + + pub async fn unlock_async(&self, path: &str, password: &str) -> Result<(), notebook::Error> { let Ok(Some(notebook)) = &self.notebook else { return Err(io::Error::from(io::ErrorKind::NotFound).into()); }; - let key = notebook.unlock(path, password)?; + let key = notebook.unlock_async(path, password).await?; self.keep_key(path, key); Ok(()) } diff --git a/crates/snowbound/src/main.rs b/crates/snowbound/src/main.rs index 7fcd736bbbab151d665760eb47247c9782db70fb..9865ad77c12cb2278db3ecde99452ce262f81631 100644 --- a/crates/snowbound/src/main.rs +++ b/crates/snowbound/src/main.rs @@ -3582,9 +3582,14 @@ impl State { } Command::DeleteVersion { page, version } => { let session = self.session.as_mut().ok_or("No section is open")?; - session.section.delete_pages(&[version])?; - session.refresh_conflicts()?; - self.commands.push(Command::OpenPage(page)); + let operation = session.section.delete_pages_async(&[version]); + self.section_edit(operation, move |state, _| { + if let Some(session) = &mut state.session { + session.refresh_conflicts()?; + } + state.commands.push(Command::OpenPage(page)); + Ok(()) + })?; } Command::CopyVersion { version, tab } => { let session = self.session.as_mut().ok_or("No section is open")?; @@ -3595,18 +3600,25 @@ impl State { _ => session.section.page(version)?, }; if tab == session.tab { - session.section.import_page(&page, &self.author)?; - session.pages = session.section.pages()?; - self.edited(Vec::new()); + let operation = session.section.import_page_async(&page, &self.author); + self.section_edit(operation, |state, _| { + if let Some(session) = &mut state.session { + session.pages = session.section.pages()?; + } + state.edited(Vec::new()); + Ok(()) + })?; } else { let library = Arc::clone(&session.library); let path = session.tabs[tab].path.clone(); let (notify, author) = (notify(self.proxy.clone()), self.author.clone()); - crate::spawn(move || { - let copied = library.open(&path, notify).and_then(|section| { - section.import_page(&page, &author)?; - Ok(section.close()?) - }); + crate::spawn_async(move || async move { + let copied: Result<(), Box> = async { + let section = library.open(&path, notify)?; + section.import_page_async(&page, &author).await?; + Ok(section.close_async().await?) + } + .await; if let Err(error) = copied { eprintln!("Copying the page failed: {error}"); } @@ -3676,14 +3688,21 @@ impl State { /// Runs `read` on a thread of its own, which lays out the page it read; `open_loaded` /// shows it unless a newer read was asked for meanwhile. fn load(&mut self, read: impl FnOnce() -> Result> + Send + 'static) { + self.load_async(move || async move { read() }); + } + + fn load_async>> + 'static>( + &mut self, + read: impl FnOnce() -> F + Send + 'static, + ) { self.loading += 1; let (id, sender, redraw) = (self.loading, self.loads.0.clone(), self.redraw.clone()); let layouts = Arc::clone(&self.layouts); let requested = Instant::now(); let scenes = Arc::clone(&self.prefetch.scenes); let open = self.session.as_ref().map(Session::key); - crate::spawn(move || { - let laid = read().and_then(|loaded| { + crate::spawn_async(move || async move { + let laid = read().await.and_then(|loaded| { lap("switch read", requested); let (shown, page) = match loaded { Loaded::Section(session, page) => (Shown::Section(session), page), @@ -3730,6 +3749,44 @@ impl State { }); } + fn complete_async + Send + 'static>( + &mut self, + operation: F, + then: impl FnOnce(&mut State, T) -> Result<(), Box> + Send + 'static, + ) -> Result<(), Box> { + #[cfg(not(target_arch = "wasm32"))] + return then(self, pollster::block_on(operation)); + #[cfg(target_arch = "wasm32")] + { + let reply = self.reply(then); + crate::spawn_async(move || async move { reply.send(operation.await) }); + Ok(()) + } + } + + fn section_edit< + T: Send + 'static, + F: Future> + Send + 'static, + >( + &mut self, + operation: F, + then: impl FnOnce(&mut State, T) -> Result<(), Box> + Send + 'static, + ) -> Result<(), Box> { + let key = self.session.as_ref().ok_or("No section is open")?.key(); + self.complete_async(operation, move |state, result| { + let value = result?; + if state + .session + .as_ref() + .is_some_and(|session| session.key() == key) + { + then(state, value) + } else { + Ok(()) + } + }) + } + /// Shows the newest page read once the pictures it shows first are drawn, or after /// `HOLD`. fn open_loaded(&mut self) -> Result<(), Box> { @@ -6207,7 +6264,6 @@ fn spawn(work: impl FnOnce() + Send + 'static) { } /// Runs asynchronous work beside the frame, creating the future on its executor. -#[cfg(feature = "live")] fn spawn_async + 'static>(work: impl FnOnce() -> F + Send + 'static) { RUNNING.fetch_add(1, Ordering::Relaxed); #[cfg(not(target_arch = "wasm32"))] diff --git a/crates/snowbound/src/manage.rs b/crates/snowbound/src/manage.rs index d0a003b7241c284485c25da2d44afd6ee7f4228d..f51c83b3331f991cc1aa66743cee020eace66145 100644 --- a/crates/snowbound/src/manage.rs +++ b/crates/snowbound/src/manage.rs @@ -781,9 +781,9 @@ impl State { .filter(|session| Arc::ptr_eq(&session.library, &library)) .map(|session| session.tabs[session.tab].path.clone()); let notify = notify(self.proxy.clone()); - self.load(move || { + self.load_async(move || async move { // A section kept open would hold a file the change moves. - library.close_kept(); + library.close_kept_async().await?; let mut notebook = library.reopen()?; let names = |notebook: &Notebook, folder: &str| -> Vec { let found = crate::library::folders(notebook.catalog(), |_| true) diff --git a/crates/snowbound/src/protection.rs b/crates/snowbound/src/protection.rs index d531850126a86969bb8e97c5670b9691551618eb..28f6dd93fe9112d9bc1781cefe0718119bce527a 100644 --- a/crates/snowbound/src/protection.rs +++ b/crates/snowbound/src/protection.rs @@ -565,23 +565,9 @@ impl State { (Some("Set Password"), _) => Some(Dialog::Set(Default::default(), None)), (Some("Change Password"), _) => Some(Dialog::Change(Default::default(), None)), (Some("Remove Password"), _) => Some(Dialog::Remove(Zeroizing::default(), false)), - (Some(_), Dialog::Unlock(value, _)) => match library.unlock(&path, &value) { - Ok(()) => { - self.commands - .push(Command::OpenSection(Arc::clone(&library), path.clone())); - None - } - Err(notebook::Error::Protected(_)) => { - Some(Dialog::Unlock(Zeroizing::default(), true)) - } - Err(error) => { - crate::platform::alert( - "Couldn't unlock the section", - &crate::plain(&error, "section"), - ); - None - } - }, + (Some(_), dialog @ Dialog::Unlock(..)) => { + return self.submit_password(library, path, dialog); + } (Some(_), Dialog::Set([value, confirm], _)) => { if value.is_empty() { Some(Dialog::Set(Default::default(), Some("Enter a password."))) @@ -607,30 +593,16 @@ impl State { Some("The new passwords don\u{2019}t match."), )) } else { - match current(&library, &path, &old) { - Some(key) => { - self.set_password( - library.clone(), - path.clone(), - Some(key), - Some(value), - ); - None - } - None => Some(Dialog::Change( - Default::default(), - Some("Password is incorrect."), - )), - } + return self.submit_password( + library, + path, + Dialog::Change([old, value, confirm], None), + ); } } - (Some(_), Dialog::Remove(value, _)) => match current(&library, &path, &value) { - Some(key) => { - self.set_password(library.clone(), path.clone(), Some(key), None); - None - } - None => Some(Dialog::Remove(Zeroizing::default(), true)), - }, + (Some(_), dialog @ Dialog::Remove(..)) => { + return self.submit_password(library, path, dialog); + } (Some(_), Dialog::Protection) => None, }; match next { @@ -640,6 +612,84 @@ impl State { Ok(()) } + fn submit_password( + &mut self, + library: Arc, + path: String, + dialog: Dialog, + ) -> Result<(), Box> { + #[cfg(target_arch = "wasm32")] + self.ui.close_popup(id()); + let operation = async move { + let result = { + let password = match &dialog { + Dialog::Unlock(value, _) | Dialog::Remove(value, _) => value, + Dialog::Change([old, ..], _) => old, + _ => unreachable!("Only password checks are submitted"), + }; + #[cfg(not(target_arch = "wasm32"))] + let result = library.unlock(&path, password); + #[cfg(target_arch = "wasm32")] + let result = library.unlock_async(&path, password).await; + result + }; + (library, path, dialog, result) + }; + self.complete_async(operation, |state, (library, path, dialog, result)| { + if let Err(error) = &result + && !matches!(error, notebook::Error::Protected(_)) + { + state.ui.close_popup(id()); + crate::platform::alert( + "Couldn't unlock the section", + &crate::plain(error, "section"), + ); + return Ok(()); + } + let key = result.as_ref().ok().and_then(|()| library.unlocked(&path)); + let next = match dialog { + Dialog::Unlock(..) => { + if result.is_ok() { + state + .commands + .push(Command::OpenSection(Arc::clone(&library), path.clone())); + None + } else { + Some(Dialog::Unlock(Zeroizing::default(), true)) + } + } + Dialog::Change([_, value, _], _) => match key { + Some(key) => { + state.set_password( + Arc::clone(&library), + path.clone(), + Some(key), + Some(value), + ); + None + } + None => Some(Dialog::Change( + Default::default(), + Some("Password is incorrect."), + )), + }, + Dialog::Remove(..) => match key { + Some(key) => { + state.set_password(Arc::clone(&library), path.clone(), Some(key), None); + None + } + None => Some(Dialog::Remove(Zeroizing::default(), true)), + }, + _ => unreachable!("Only password checks are submitted"), + }; + match next { + Some(dialog) => state.ask_password(library, path, dialog), + None => state.ui.close_popup(id()), + } + Ok(()) + }) + } + /// Sets, changes or removes the password of `path` in `library`, closing its section /// first: the section is written anew, and its replica goes with the old file. fn set_password( @@ -649,38 +699,113 @@ impl State { key: Option, password: Option>, ) { - let open = self.session.as_ref().is_some_and(|session| { - Arc::ptr_eq(&session.library, &library) && session.tabs[session.tab].path == path - }); - if open { - let closed = self.persist().and_then(|()| { - if let Some(session) = self.session.take() { - session.section.close()?; + let replica = self + .session + .as_ref() + .filter(|session| { + Arc::ptr_eq(&session.library, &library) && session.tabs[session.tab].path == path + }) + .map(|session| Arc::clone(session.section.replica())); + let Some(replica) = replica else { + library.lock(|locked, _| locked == path); + self.restructure( + library, + Structure::Password { + path, + key, + password, + }, + ); + return; + }; + if let Err(error) = self.persist() { + return crate::platform::alert( + "Couldn't set the password", + &crate::plain(&*error, "section"), + ); + } + let operation = async move { + let through = replica.pending()?.last().map_or(0, |edit| edit.id); + replica.written_async().await?; + Ok::<_, notebook::Error>((replica, through)) + }; + let _ = self.complete_async(operation, move |state, result| { + let prepared = (|| -> Result<(), Box> { + let (replica, through) = result?; + let open = state + .session + .as_ref() + .is_some_and(|session| Arc::ptr_eq(session.section.replica(), &replica)); + if open { + state.persist()?; + let pending = replica.pending()?.last().map_or(0, |edit| edit.id); + if pending > through { + state.set_password(library, path, key, password); + return Ok(()); + } } - Ok(()) - }); - if let Err(error) = closed { - return crate::platform::alert( + let stopping = state + .session + .as_mut() + .filter(|_| open) + .map(|session| session.section.stop_async()); + let operation = async move { + if let Some(stopping) = stopping { + stopping.await?; + } + let through = replica.pending()?.last().map_or(0, |edit| edit.id); + replica.written_async().await?; + Ok::<_, notebook::Error>((replica, through)) + }; + state.complete_async(operation, move |state, result| { + let closed = + (|| -> Result<(), Box> { + let (replica, through) = result?; + if state.session.as_ref().is_some_and(|session| { + Arc::ptr_eq(session.section.replica(), &replica) + }) { + state.persist()?; + if replica + .pending()? + .last() + .is_some_and(|edit| edit.id > through) + { + state.set_password(library, path, key, password); + return Ok(()); + } + if let Some(session) = state.session.take() { + session.section.close()?; + } + } + drop(replica); + state.prefetch.forget(&library.key(&path)); + library.lock(|locked, _| locked == path); + state.restructure( + library, + Structure::Password { + path, + key, + password, + }, + ); + Ok(()) + })(); + if let Err(error) = closed { + crate::platform::alert( + "Couldn't set the password", + &crate::plain(&*error, "section"), + ); + } + Ok(()) + }) + })(); + if let Err(error) = prepared { + crate::platform::alert( "Couldn't set the password", &crate::plain(&*error, "section"), ); } - self.prefetch.forget(&library.key(&path)); - } - library.lock(|locked, _| locked == path); - self.restructure( - library, - Structure::Password { - path, - key, - password, - }, - ); + Ok(()) + }); } } - -/// The key `password` opens the section at `path` with, if it does. -fn current(library: &Library, path: &str, password: &str) -> Option { - library.unlock(path, password).ok()?; - library.unlocked(path) -} diff --git a/crates/snowbound/src/recycle.rs b/crates/snowbound/src/recycle.rs index bf0f942e8e7ad9f872772ff46c9eedd48cbc247f..0415a8fbacff60f0768c06964c72813fa9410e75 100644 --- a/crates/snowbound/src/recycle.rs +++ b/crates/snowbound/src/recycle.rs @@ -135,8 +135,8 @@ impl State { (spaces.clone().skip(at).find(kept)).or_else(|| spaces.take(at).rev().find(kept)) }); let (author, proxy) = (self.author.clone(), self.proxy.clone()); - self.load(move || { - carry_out(&library, &replica, request, &author)?; + self.load_async(move || async move { + carry_out(&library, &replica, request, &author).await?; if let Some(show) = if gone.contains(&shown) { after } else { @@ -192,7 +192,7 @@ impl State { /// Does `request` to the bin's section `replica` holds, in `library`, as `author`: a page /// restored goes to its section as one op, keeping its identity, a copy under a new one, and /// what leaves the bin goes from it as another. -fn carry_out( +async fn carry_out( library: &Library, replica: ¬ebook::Replica, request: Request, @@ -203,31 +203,36 @@ fn carry_out( let page = replica.page(space)?; let section = library.open(&path, || {})?; if copy { - section.import_page(&page, author)?; + section.import_page_async(&page, author).await?; } else { let ops = vec![notebook::session::moved(&page, author)?]; - section.replica().apply( - author, - Edit { - at: filetime(), - ops, - }, - )?; + section + .replica() + .apply_async( + author, + Edit { + at: filetime(), + ops, + }, + ) + .await?; } - section.close()?; + section.close_async().await?; (!copy).then_some(vec![space]).unwrap_or_default() } Request::Purge(spaces) => spaces, }; if !gone.is_empty() { let ops = vec![Op::Section(SectionOp::Delete(gone))]; - replica.apply( - author, - Edit { - at: filetime(), - ops, - }, - )?; + replica + .apply_async( + author, + Edit { + at: filetime(), + ops, + }, + ) + .await?; } Ok(()) } @@ -360,14 +365,26 @@ mod tests { path: FIRST.into(), copy, }; - carry_out(&library, bin.replica(), restore(back, false), AUTHOR).unwrap(); - carry_out(&library, bin.replica(), restore(copied, true), AUTHOR).unwrap(); - carry_out( + pollster::block_on(carry_out( + &library, + bin.replica(), + restore(back, false), + AUTHOR, + )) + .unwrap(); + pollster::block_on(carry_out( + &library, + bin.replica(), + restore(copied, true), + AUTHOR, + )) + .unwrap(); + pollster::block_on(carry_out( &library, bin.replica(), Request::Purge(vec![purged]), AUTHOR, - ) + )) .unwrap(); bin.close().unwrap(); published(&library, FIRST); diff --git a/crates/snowbound/src/undo.rs b/crates/snowbound/src/undo.rs index c040325a8d39d62d57e5a3a3af72c1314a41c2de..21def366923867254053c50a6a72817dc37944c0 100644 --- a/crates/snowbound/src/undo.rs +++ b/crates/snowbound/src/undo.rs @@ -350,14 +350,16 @@ struct Applied { } impl Site { - fn apply(&self, ops: Vec) -> Result<(), Box> { - self.replica.apply( - &self.author, - Edit { - at: crate::filetime(), - ops, - }, - )?; + async fn apply(&self, ops: Vec) -> Result<(), Box> { + self.replica + .apply_async( + &self.author, + Edit { + at: crate::filetime(), + ops, + }, + ) + .await?; Ok(()) } @@ -408,7 +410,11 @@ impl Site { /// Does `change` to the section's pages, `shown` the page open; `None` where what it /// changes is gone or holds what it did not make, which it leaves alone. - fn change(&self, change: Change, shown: ExGuid) -> Result, Box> { + async fn change( + &self, + change: Change, + shown: ExGuid, + ) -> Result, Box> { let listed = self.replica.pages()?; let at = |space: ExGuid| listed.iter().position(|(listed, ..)| *listed == space); let binned = matches!(self.library.notebook, Ok(Some(_))); @@ -449,8 +455,9 @@ impl Site { if !created && binned { let pages: Vec = kept.iter().map(|kept| kept.page.clone()).collect(); self.library.reopen()?.recycle_pages(&pages, &self.author)?; + self.replica.written_async().await?; } - self.apply(ops)?; + self.apply(ops).await?; Applied { undo: Change::Restore { pages: kept, @@ -494,7 +501,7 @@ impl Site { spaces.push(space); } let gone = self.take_added(added, &listed, &mut ops); - self.apply(ops)?; + self.apply(ops).await?; if !created && binned { let identities: Vec<[u8; 16]> = back.iter().filter_map(|kept| kept.page.identity).collect(); @@ -515,7 +522,8 @@ impl Site { let Some(show) = edits.first().map(PageEdit::space) else { return Ok(None); }; - self.apply(vec![Op::Section(SectionOp::Pages(edits))])?; + self.apply(vec![Op::Section(SectionOp::Pages(edits))]) + .await?; Applied { undo: Change::Arrange { order: listed @@ -533,7 +541,7 @@ impl Site { return Ok(None); } let op = crate::rename::retitled(&self.replica.page(page)?, page, to.clone())?; - self.apply(vec![op])?; + self.apply(vec![op]).await?; Applied { undo: Change::Retitle { page, @@ -555,15 +563,18 @@ impl Site { let mut ops = vec![Op::Section(SectionOp::Delete(vec![page]))]; let (show, added) = self.after(&listed, &[page], None, &mut ops); let section = self.library.open(&path, || {})?; - section.replica().apply( - &self.author, - Edit { - at: crate::filetime(), - ops: vec![import], - }, - )?; - section.close()?; - self.apply(ops)?; + section + .replica() + .apply_async( + &self.author, + Edit { + at: crate::filetime(), + ops: vec![import], + }, + ) + .await?; + section.close_async().await?; + self.apply(ops).await?; Applied { undo: Change::Return { page: arrived, @@ -592,7 +603,7 @@ impl Site { false => None, }; let Some(sent) = sent else { - section.close()?; + section.close_async().await?; return Ok(None); }; let import = notebook::session::moved(&sent, &self.author)?; @@ -606,15 +617,18 @@ impl Site { )?])), ]; let gone = self.take_added(added, &listed, &mut ops); - self.apply(ops)?; - section.replica().apply( - &self.author, - Edit { - at: crate::filetime(), - ops: vec![Op::Section(SectionOp::Delete(vec![page]))], - }, - )?; - section.close()?; + self.apply(ops).await?; + section + .replica() + .apply_async( + &self.author, + Edit { + at: crate::filetime(), + ops: vec![Op::Section(SectionOp::Delete(vec![page]))], + }, + ) + .await?; + section.close_async().await?; Applied { undo: Change::Send { page: back, target }, show: back, @@ -1052,8 +1066,8 @@ impl State { let proxy = self.proxy.clone(); self.undo.busy += 1; self.switching = Some((None, web_time::Instant::now())); - self.load(move || { - let applied = site.change(change, shown); + self.load_async(move || async move { + let applied = site.change(change, shown).await; let show = match &applied { Ok(Some(applied)) => applied.show, _ => shown, @@ -1382,7 +1396,7 @@ mod tests { /// Does `change`, as the app's thread does, returning what takes it back. fn change(&self, change: Change) -> Option { let shown = self.pages()[0].0; - self.site().change(change, shown).unwrap() + pollster::block_on(self.site().change(change, shown)).unwrap() } } diff --git a/crates/snowbound/src/web.rs b/crates/snowbound/src/web.rs index c8e93c2fcf1d4c7bd5c67986b9692e94082bcdb7..0e171bbe17a798bc26560b39bb75606c031e3d60 100644 --- a/crates/snowbound/src/web.rs +++ b/crates/snowbound/src/web.rs @@ -1054,7 +1054,8 @@ fn restore(files: js_sys::Array) { } /// Writes the files changed since the last time out to the browser's storage. -fn store() { +#[wasm_bindgen] +pub fn flush_storage() { use notebook::fs::Change; STORE_DUE.set(None); let changes = js_sys::Array::new(); @@ -1489,7 +1490,7 @@ pub fn flush() { report(error); } }); - store(); + flush_storage(); } #[wasm_bindgen] @@ -1585,7 +1586,7 @@ fn turn(state: &mut State) -> Result<(), Box> { STORE_DUE.set(Some(now + STORE_AFTER)); } if STORE_DUE.get().is_some_and(|due| due <= now) { - store(); + flush_storage(); } let updating = follow_updates(state, now); let next = [ @@ -1703,7 +1704,7 @@ pub fn update_now(state: &mut State) { if let Err(error) = state.persist() { report(error); } - store(); + flush_storage(); let Some(build) = UPDATE.with_borrow(Clone::clone) else { return RELOADING.set(false); }; diff --git a/crates/snowbound/web/glue.js b/crates/snowbound/web/glue.js index 7a49576deb8979e9b4f90c85931f6597223539a0..420770abc801f655a5cfa1c99a59345bdabb1364 100644 --- a/crates/snowbound/web/glue.js +++ b/crates/snowbound/web/glue.js @@ -639,7 +639,7 @@ export function adoptCanvas(fresh) { /** Wires the page's canvas, text area and file input to `module`'s exports. */ export function attach(module) { wasm = module; - globalThis.snowboundFlushStorage = () => { wasm.flush(); return settled(); }; + globalThis.snowboundFlushStorage = () => { wasm.flush_storage(); return settled(); }; canvas = document.getElementById("page"); input = document.getElementById("input"); picker = document.getElementById("files"); diff --git a/tools/web/test_storage_barrier.mjs b/tools/web/test_storage_barrier.mjs new file mode 100644 index 0000000000000000000000000000000000000000..62f6c22d70e54024d6e31ad3bbbc621455b6e9a9 --- /dev/null +++ b/tools/web/test_storage_barrier.mjs @@ -0,0 +1,59 @@ +import assert from 'node:assert/strict'; +import { readFile } from 'node:fs/promises'; +import test from 'node:test'; +import vm from 'node:vm'; + +test('the durability bridge drains files without reentering the editor and waits for ordered worker receipts', async () => { + const source = await readFile(new URL('../../crates/snowbound/web/glue.js', import.meta.url), 'utf8'); + const messages = []; + let worker; + const element = { style: {}, addEventListener() {}, focus() {}, remove() {} }; + const context = vm.createContext({ + Worker: class { + constructor() { worker = this; } + postMessage(data) { + messages.push(data); + if (data.kind === 'load') queueMicrotask(() => this.onmessage({ data: [] })); + } + }, + Blob, + URL, + document: { getElementById: () => element, addEventListener() {} }, + localStorage: { getItem: () => null }, + ResizeObserver: class { observe() {} }, + matchMedia: () => ({ addEventListener() {} }), + addEventListener() {}, + console, + }); + vm.runInContext(source.replace(/\bexport /g, '').replaceAll('import.meta.url', JSON.stringify(import.meta.url)), context); + await vm.runInContext('loadFiles()', context); + const flushed = []; + context.module = { + flush() { throw Error('Durability reentered State.persist'); }, + flush_storage() { + const marker = flushed.length + 1; + flushed.push(marker); + context.nextChange = [['replica-wal', marker, [[0, new Uint8Array([marker])]]]]; + vm.runInContext('storeFiles(nextChange)', context); + }, + }; + vm.runInContext('attach(module)', context); + const replies = []; + const first = vm.runInContext('snowboundFlushStorage()', context).then(() => replies.push(1)); + const second = vm.runInContext('snowboundFlushStorage()', context).then(() => replies.push(2)); + assert.deepEqual(messages.map(message => message.kind), ['load', 'store', 'settle', 'store', 'settle']); + await Promise.resolve(); + assert.deepEqual(replies, []); + worker.onmessage({ data: { error: null } }); + await first; + assert.deepEqual(replies, [1]); + worker.onmessage({ data: { error: null } }); + await second; + assert.deepEqual(replies, [1, 2]); + const rejected = vm.runInContext('snowboundFlushStorage()', context); + worker.onmessage({ data: { error: 'Storage unavailable' } }); + await assert.rejects(rejected, /Storage unavailable/); + const retried = vm.runInContext('snowboundFlushStorage()', context); + worker.onmessage({ data: { error: null } }); + await retried; +}); -- 2.54.0