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; +});