authorclover caruso <git@paperclover.net> 2026-10-06 00:26:55-07:00
committerclover caruso <git@paperclover.net> 2026-10-10 22:56:05-07:00
logfe5ea9c240ad4fc12520c83370d4938f97ff87c5
tree62f15ebb1829416bdba1e85d4f85cf97952687a4
parent9a109461103375723db50473c2562c7195eea36c
signature Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU (~clover)

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

17 files changed, 936 insertions(+), 283 deletions(-)

crates/notebook/src/background.rs+1-1
...@@ -433,7 +433,7 @@ impl Background {...@@ -433,7 +433,7 @@ impl Background {
433 }433 }
434 }434 }
435 let (shared, path) = (Arc::downgrade(&self.0), path.to_owned());435 let (shared, path) = (Arc::downgrade(&self.0), path.to_owned());
436 if let Ok(mut released) = replica.released.0.lock() {436 if let Ok(mut released) = replica.section.released.0.lock() {
437 *released = Some(Box::new(move || {437 *released = Some(Box::new(move || {
438 if let Some(shared) = shared.upgrade() {438 if let Some(shared) = shared.upgrade() {
439 shared.released(&path);439 shared.released(&path);
crates/notebook/src/lib.rs+40-10
...@@ -112,8 +112,6 @@ pub struct Replica {...@@ -112,8 +112,6 @@ pub struct Replica {
112 synchronization: Mutex<()>,112 synchronization: Mutex<()>,
113 /// The section's root object space, which names the document.113 /// The section's root object space, which names the document.
114 root: ExGuid,114 root: ExGuid,
115 /// Last, so that it runs once the fields above have closed the cache.
116 released: Released,
117}115}
118116
119/// Runs once the replica is released and its cache closed, as `Background::hold` asks.117/// Runs once the replica is released and its cache closed, as `Background::hold` asks.
...@@ -254,7 +252,6 @@ impl Replica {...@@ -254,7 +252,6 @@ impl Replica {
254 section,252 section,
255 synchronization: Mutex::new(()),253 synchronization: Mutex::new(()),
256 root,254 root,
257 released: Released::default(),
258 })255 })
259 }256 }
260257
...@@ -281,18 +278,39 @@ impl Replica {...@@ -281,18 +278,39 @@ impl Replica {
281 .map_err(|_| io::Error::other("The section thread stopped"))?278 .map_err(|_| io::Error::other("The section thread stopped"))?
282 }279 }
283280
281 async fn ask_async<T: Send + 'static>(
282 &self,
283 request: impl FnOnce(working::Reply<T>) -> working::Request + Send,
284 ) -> Result<T> {
285 let (sender, receiver) = task::response();
286 self.send(request(Box::new(move |result| {
287 let _ = sender.send(result);
288 })))?;
289 #[cfg(target_arch = "wasm32")]
290 let answer = task::receive(&receiver).await;
291 #[cfg(not(target_arch = "wasm32"))]
292 let answer = receiver.recv();
293 answer.map_err(|_| io::Error::other("The section thread stopped"))?
294 }
295
284 /// Applies an edit and queues it durably for publication, returning its id once296 /// Applies an edit and queues it durably for publication, returning its id once
285 /// written. A refused edit returns `Rejected` and leaves every page as it was.297 /// written. A refused edit returns `Rejected` and leaves every page as it was.
298 /// Browser callers use `apply_async`; this returns `WouldBlock` without submitting.
286 pub fn apply(&self, author: &str, edit: Edit) -> Result<u64> {299 pub fn apply(&self, author: &str, edit: Edit) -> Result<u64> {
287 self.ask(|reply| working::Request::Apply {300 task::blocking(self.apply_async(author, edit))?
301 }
302
303 /// Applies an edit, awaiting its durable local queue receipt without blocking browser reads.
304 pub async fn apply_async(&self, author: &str, edit: Edit) -> Result<u64> {
305 self.ask_async(|reply| working::Request::Apply {
288 author: author.to_owned(),306 author: author.to_owned(),
289 edit,307 edit,
290 reply,308 reply,
291 })309 })
310 .await
292 }311 }
293312
294 /// `apply` without waiting: `reply` runs on the section thread once the edit is313 /// `apply` without waiting: `reply` runs once the edit is durable or refused.
295 /// durable or refused.
296 pub(crate) fn submit(314 pub(crate) fn submit(
297 &self,315 &self,
298 author: &str,316 author: &str,
...@@ -338,14 +356,26 @@ impl Replica {...@@ -338,14 +356,26 @@ impl Replica {
338 /// The section image the queued edits leave, the unsealed ones sealed as one more356 /// The section image the queued edits leave, the unsealed ones sealed as one more
339 /// revision whose identities differ per call: O(section).357 /// revision whose identities differ per call: O(section).
340 pub fn snapshot(&self) -> Result<Vec<u8>> {358 pub fn snapshot(&self) -> Result<Vec<u8>> {
341 self.written()?;359 self.ask(|reply| working::Request::Flush {
360 reply,
361 durable: false,
362 })?;
342 working::image(&*self.lock()?, self.section.key.as_ref())363 working::image(&*self.lock()?, self.section.key.as_ref())
343 }364 }
344365
345 /// Waits until the edits applied before it are written to the queue, as reads answer366 /// Waits until the edits applied before it are durable in the local queue.
346 /// before an open burst of edits is.367 /// Browser callers use `written_async`; this returns `WouldBlock` without submitting.
347 pub fn written(&self) -> Result<()> {368 pub fn written(&self) -> Result<()> {
348 self.ask(|reply| working::Request::Flush { reply })369 task::blocking(self.written_async())?
370 }
371
372 /// Waits for durable local storage of the edits submitted before this request.
373 pub async fn written_async(&self) -> Result<()> {
374 self.ask_async(|reply| working::Request::Flush {
375 reply,
376 durable: true,
377 })
378 .await
349 }379 }
350380
351 /// The section file's identity, which internal links name as `section-id`.381 /// The section file's identity, which internal links name as `section-id`.
crates/notebook/src/session.rs+122-43
...@@ -1347,6 +1347,10 @@ impl Notebook {...@@ -1347,6 +1347,10 @@ impl Notebook {
1347 /// queued again under the key: each page they changed comes back as a copy, as a page1347 /// queued again under the key: each page they changed comes back as a copy, as a page
1348 /// another client removed does, the section's pages having taken new identities.1348 /// another client removed does, the section's pages having taken new identities.
1349 pub fn unlock(&self, path: &str, password: &str) -> Result<protected::Key> {1349 pub fn unlock(&self, path: &str, password: &str) -> Result<protected::Key> {
1350 crate::task::blocking(self.unlock_async(path, password))?
1351 }
1352
1353 pub async fn unlock_async(&self, path: &str, password: &str) -> Result<protected::Key> {
1350 let section = self.section_path(path)?;1354 let section = self.section_path(path)?;
1351 let image = self.storage.read(&section.path)?;1355 let image = self.storage.read(&section.path)?;
1352 let key = protected::Key::open(&image, password)?;1356 let key = protected::Key::open(&image, password)?;
...@@ -1365,7 +1369,7 @@ impl Notebook {...@@ -1365,7 +1369,7 @@ impl Notebook {
1365 continue;1369 continue;
1366 };1370 };
1367 for (author, edit) in old.copies()? {1371 for (author, edit) in old.copies()? {
1368 replica.apply(&author, edit)?;1372 replica.apply_async(&author, edit).await?;
1369 }1373 }
1370 drop(old);1374 drop(old);
1371 remove_replica(&path)?;1375 remove_replica(&path)?;
...@@ -2242,31 +2246,60 @@ impl Section {...@@ -2242,31 +2246,60 @@ impl Section {
2242 /// fresh identities, queued like the user's own edits. Returns the new page's space.2246 /// fresh identities, queued like the user's own edits. Returns the new page's space.
2243 /// Content outside the model refuses to copy.2247 /// Content outside the model refuses to copy.
2244 pub fn import_page(&self, page: &Page, author: &str) -> Result<ExGuid> {2248 pub fn import_page(&self, page: &Page, author: &str) -> Result<ExGuid> {
2245 let creation = PageCreation::new(None, Some(&page.title), author)?;2249 crate::task::blocking(self.import_page_async(page, author))?
2246 let space = creation.space();2250 }
2247 self.replica.apply(2251
2248 author,2252 pub fn import_page_async(
2249 Edit {2253 &self,
2250 at: crate::now(),2254 page: &Page,
2251 ops: vec![Op::Section(SectionOp::Import {2255 author: &str,
2252 creation,2256 ) -> impl Future<Output = Result<ExGuid>> + Send + 'static + use<> {
2253 page: page.copy()?,2257 let replica = Arc::clone(&self.replica);
2254 })],2258 let author = author.to_owned();
2255 },2259 let creation = PageCreation::new(None, Some(&page.title), &author);
2256 )?;2260 let page = page.copy();
2257 Ok(space)2261 async move {
2262 let creation = creation?;
2263 let space = creation.space();
2264 replica
2265 .apply_async(
2266 &author,
2267 Edit {
2268 at: crate::now(),
2269 ops: vec![Op::Section(SectionOp::Import {
2270 creation,
2271 page: page?,
2272 })],
2273 },
2274 )
2275 .await?;
2276 Ok(space)
2277 }
2258 }2278 }
22592279
2260 /// Removes pages or conflict pages permanently, queued like the user's own edits (a2280 /// Removes pages or conflict pages permanently, queued like the user's own edits (a
2261 /// move across sections is `import_page` there, then this here).2281 /// move across sections is `import_page` there, then this here).
2262 pub fn delete_pages(&self, pages: &[ExGuid]) -> Result<u64> {2282 pub fn delete_pages(&self, pages: &[ExGuid]) -> Result<u64> {
2263 self.replica.apply(2283 crate::task::blocking(self.delete_pages_async(pages))?
2264 "",2284 }
2265 Edit {2285
2266 at: crate::now(),2286 pub fn delete_pages_async(
2267 ops: vec![Op::Section(SectionOp::Delete(pages.to_vec()))],2287 &self,
2268 },2288 pages: &[ExGuid],
2269 )2289 ) -> impl Future<Output = Result<u64>> + Send + 'static + use<> {
2290 let replica = Arc::clone(&self.replica);
2291 let pages = pages.to_vec();
2292 async move {
2293 replica
2294 .apply_async(
2295 "",
2296 Edit {
2297 at: crate::now(),
2298 ops: vec![Op::Section(SectionOp::Delete(pages))],
2299 },
2300 )
2301 .await
2302 }
2270 }2303 }
22712304
2272 pub fn status(&self, id: u64) -> Result<Option<EditStatus>> {2305 pub fn status(&self, id: u64) -> Result<Option<EditStatus>> {
...@@ -2302,32 +2335,61 @@ impl Section {...@@ -2302,32 +2335,61 @@ impl Section {
2302 /// Makes a page's version its current state, the page as it stood becoming the newest2335 /// Makes a page's version its current state, the page as it stood becoming the newest
2303 /// version, queued like the user's own edits.2336 /// version, queued like the user's own edits.
2304 pub fn restore_version(&self, space: ExGuid, version: ExGuid, author: &str) -> Result<u64> {2337 pub fn restore_version(&self, space: ExGuid, version: ExGuid, author: &str) -> Result<u64> {
2305 self.replica.apply(2338 crate::task::blocking(self.restore_version_async(space, version, author))?
2306 author,2339 }
2307 Edit {2340
2308 at: crate::now(),2341 pub fn restore_version_async(
2309 ops: vec![Op::Section(SectionOp::restore(space, version)?)],2342 &self,
2310 },2343 space: ExGuid,
2311 )2344 version: ExGuid,
2345 author: &str,
2346 ) -> impl Future<Output = Result<u64>> + Send + 'static + use<> {
2347 let replica = Arc::clone(&self.replica);
2348 let author = author.to_owned();
2349 let op = SectionOp::restore(space, version);
2350 async move {
2351 replica
2352 .apply_async(
2353 &author,
2354 Edit {
2355 at: crate::now(),
2356 ops: vec![Op::Section(op?)],
2357 },
2358 )
2359 .await
2360 }
2312 }2361 }
23132362
2314 /// Deletes versions of pages, queued like the user's own edits.2363 /// Deletes versions of pages, queued like the user's own edits.
2315 pub fn delete_versions(&self, versions: &[(ExGuid, Vec<ExGuid>)]) -> Result<u64> {2364 pub fn delete_versions(&self, versions: &[(ExGuid, Vec<ExGuid>)]) -> Result<u64> {
2316 self.replica.apply(2365 crate::task::blocking(self.delete_versions_async(versions))?
2317 "",2366 }
2318 Edit {2367
2319 at: crate::now(),2368 pub fn delete_versions_async(
2320 ops: versions2369 &self,
2321 .iter()2370 versions: &[(ExGuid, Vec<ExGuid>)],
2322 .map(|(page, versions)| {2371 ) -> impl Future<Output = Result<u64>> + Send + 'static + use<> {
2323 Op::Section(SectionOp::DeleteVersions {2372 let replica = Arc::clone(&self.replica);
2324 page: *page,2373 let ops = versions
2325 versions: versions.clone(),2374 .iter()
2326 })2375 .map(|(page, versions)| {
2327 })2376 Op::Section(SectionOp::DeleteVersions {
2328 .collect(),2377 page: *page,
2329 },2378 versions: versions.clone(),
2330 )2379 })
2380 })
2381 .collect();
2382 async move {
2383 replica
2384 .apply_async(
2385 "",
2386 Edit {
2387 at: crate::now(),
2388 ops,
2389 },
2390 )
2391 .await
2392 }
2331 }2393 }
23322394
2333 /// Retires an uncertain attempt after review, exporting the queue to `archive` first:2395 /// Retires an uncertain attempt after review, exporting the queue to `archive` first:
...@@ -2409,6 +2471,23 @@ impl Section {...@@ -2409,6 +2471,23 @@ impl Section {
2409 None => Ok(()),2471 None => Ok(()),
2410 }2472 }
2411 }2473 }
2474
2475 /// Stops synchronization without consuming the section's recoverable model.
2476 pub fn stop_async(&mut self) -> impl Future<Output = Result<()>> + Send + 'static + use<> {
2477 let worker = self.worker.take();
2478 async move {
2479 match worker {
2480 Some(worker) => worker.stop_async().await,
2481 None => Ok(()),
2482 }
2483 }
2484 }
2485
2486 /// Makes the submitted edits locally durable and awaits synchronization before closing.
2487 pub async fn close_async(mut self) -> Result<()> {
2488 self.replica.written_async().await?;
2489 self.stop_async().await
2490 }
2412}2491}
24132492
2414/// The section file itself as the publication target, under OneNote-compatible exclusion.2493/// The section file itself as the publication target, under OneNote-compatible exclusion.
crates/notebook/src/task.rs+45-18
...@@ -10,14 +10,25 @@ pub(crate) use std::{...@@ -10,14 +10,25 @@ pub(crate) use std::{
10 thread::JoinHandle,10 thread::JoinHandle,
11};11};
1212
13#[cfg(all(feature = "live", not(target_arch = "wasm32")))]13#[cfg(not(target_arch = "wasm32"))]
14pub(crate) type Answer<T> = std::sync::mpsc::Sender<T>;14pub(crate) type Answer<T> = std::sync::mpsc::Sender<T>;
1515
16#[cfg(all(feature = "live", not(target_arch = "wasm32")))]16#[cfg(not(target_arch = "wasm32"))]
17pub(crate) fn response<T>() -> (Answer<T>, std::sync::mpsc::Receiver<T>) {17pub(crate) fn response<T>() -> (Answer<T>, std::sync::mpsc::Receiver<T>) {
18 std::sync::mpsc::channel()18 std::sync::mpsc::channel()
19}19}
2020
21/// Runs native blocking work; browser callers are refused before the future is polled.
22pub(crate) fn blocking<T>(work: impl Future<Output = T>) -> io::Result<T> {
23 #[cfg(not(target_arch = "wasm32"))]
24 return Ok(complete(work));
25 #[cfg(target_arch = "wasm32")]
26 {
27 drop(work);
28 Err(io::ErrorKind::WouldBlock.into())
29 }
30}
31
21#[cfg(all(feature = "live", not(target_arch = "wasm32")))]32#[cfg(all(feature = "live", not(target_arch = "wasm32")))]
22pub(crate) async fn answered<T>(33pub(crate) async fn answered<T>(
23 receiver: &std::sync::mpsc::Receiver<T>,34 receiver: &std::sync::mpsc::Receiver<T>,
...@@ -134,24 +145,23 @@ mod web {...@@ -134,24 +145,23 @@ mod web {
134 }145 }
135 }146 }
136147
148 #[cfg(feature = "live")]
137 pub(crate) async fn answered<T>(149 pub(crate) async fn answered<T>(
138 receiver: &Answered<T>,150 receiver: &Answered<T>,
139 timeout: Duration,151 timeout: Duration,
140 ) -> Result<T, std::sync::mpsc::RecvTimeoutError> {152 ) -> Result<T, std::sync::mpsc::RecvTimeoutError> {
141 let deadline = web_time::Instant::now() + timeout;153 let deadline = web_time::Instant::now() + timeout;
142 let mut armed = false;154 let mut armed = false;
155 let mut answer = std::pin::pin!(receive(receiver));
143 std::future::poll_fn(|context| {156 std::future::poll_fn(|context| {
144 let mut response = receiver.0.lock().unwrap();157 if let Poll::Ready(answer) = answer.as_mut().poll(context) {
145 if let Some(value) = response.value.take() {158 return Poll::Ready(
146 return Poll::Ready(Ok(value));159 answer.map_err(|_| std::sync::mpsc::RecvTimeoutError::Disconnected),
147 }160 );
148 if response.closed {
149 return Poll::Ready(Err(std::sync::mpsc::RecvTimeoutError::Disconnected));
150 }161 }
151 if web_time::Instant::now() >= deadline {162 if web_time::Instant::now() >= deadline {
152 return Poll::Ready(Err(std::sync::mpsc::RecvTimeoutError::Timeout));163 return Poll::Ready(Err(std::sync::mpsc::RecvTimeoutError::Timeout));
153 }164 }
154 response.waiting = Some(context.waker().clone());
155 if !std::mem::replace(&mut armed, true) {165 if !std::mem::replace(&mut armed, true) {
156 after(timeout, context.waker().clone());166 after(timeout, context.waker().clone());
157 }167 }
...@@ -160,6 +170,23 @@ mod web {...@@ -160,6 +170,23 @@ mod web {
160 .await170 .await
161 }171 }
162172
173 pub(crate) async fn receive<T>(
174 receiver: &Answered<T>,
175 ) -> Result<T, std::sync::mpsc::RecvError> {
176 std::future::poll_fn(|context| {
177 let mut response = receiver.0.lock().unwrap();
178 if let Some(value) = response.value.take() {
179 return Poll::Ready(Ok(value));
180 }
181 if response.closed {
182 return Poll::Ready(Err(std::sync::mpsc::RecvError));
183 }
184 response.waiting = Some(context.waker().clone());
185 Poll::Pending
186 })
187 .await
188 }
189
163 #[derive(Default)]190 #[derive(Default)]
164 struct Bell {191 struct Bell {
165 rung: bool,192 rung: bool,
...@@ -229,13 +256,17 @@ mod web {...@@ -229,13 +256,17 @@ mod web {
229 .await256 .await
230 }257 }
231258
232 /// A task the browser's event loop runs. It cannot be waited for, and ends on its own.259 /// A task the browser's event loop runs, with a wakeable completion receipt.
233 pub(crate) struct JoinHandle<T>(Arc<Mutex<Option<T>>>);260 pub(crate) struct JoinHandle<T>(Answered<T>);
234261
235 impl<T> JoinHandle<T> {262 impl<T> JoinHandle<T> {
236 /// What the task ended with, if it has ended.263 /// What the task ended with, if it has ended.
237 pub(crate) fn finished(self) -> Option<T> {264 pub(crate) fn finished(self) -> Option<T> {
238 self.0.lock().ok()?.take()265 self.0.0.lock().ok()?.value.take()
266 }
267
268 pub(crate) async fn join(self) -> Result<T, std::sync::mpsc::RecvError> {
269 receive(&self.0).await
239 }270 }
240 }271 }
241272
...@@ -243,14 +274,10 @@ mod web {...@@ -243,14 +274,10 @@ mod web {
243 _: &str,274 _: &str,
244 start: impl FnOnce() -> F,275 start: impl FnOnce() -> F,
245 ) -> io::Result<JoinHandle<T>> {276 ) -> io::Result<JoinHandle<T>> {
246 let output = Arc::new(Mutex::new(None));277 let (ended, output) = response();
247 let ended = Arc::clone(&output);
248 let work = start();278 let work = start();
249 wasm_bindgen_futures::spawn_local(async move {279 wasm_bindgen_futures::spawn_local(async move {
250 let result = work.await;280 let _ = ended.send(work.await);
251 if let Ok(mut ended) = ended.lock() {
252 *ended = Some(result);
253 }
254 });281 });
255 Ok(JoinHandle(output))282 Ok(JoinHandle(output))
256 }283 }
crates/notebook/src/worker.rs+16
...@@ -130,6 +130,22 @@ impl SyncWorker {...@@ -130,6 +130,22 @@ impl SyncWorker {
130 .join()130 .join()
131 .map_err(|_| io::Error::other("Synchronization worker panicked"))?131 .map_err(|_| io::Error::other("Synchronization worker panicked"))?
132 }132 }
133
134 pub(crate) async fn stop_async(self) -> Result<()> {
135 #[cfg(not(target_arch = "wasm32"))]
136 return self.stop();
137 #[cfg(target_arch = "wasm32")]
138 {
139 let mut worker = self;
140 worker.signal.stopped.store(true, Ordering::Release);
141 worker.signal.wake();
142 let thread = worker.thread.take().expect("Worker owns its thread");
143 thread
144 .join()
145 .await
146 .map_err(|_| io::Error::other("Synchronization worker panicked"))?
147 }
148 }
133}149}
134150
135impl Drop for SyncWorker {151impl Drop for SyncWorker {
crates/notebook/src/working.rs+137-14
...@@ -44,9 +44,10 @@ pub(crate) enum Request {...@@ -44,9 +44,10 @@ pub(crate) enum Request {
44 version: ExGuid,44 version: ExGuid,
45 reply: Reply<Page>,45 reply: Reply<Page>,
46 },46 },
47 /// Answers once the edits before it are written.47 /// Answers once the edits before it are written, optionally awaiting host durability.
48 Flush {48 Flush {
49 reply: Reply<()>,49 reply: Reply<()>,
50 durable: bool,
50 },51 },
51 /// Seals the open batch, when no sealed batch waits for publication, recording the52 /// Seals the open batch, when no sealed batch waits for publication, recording the
52 /// attempt to publish it.53 /// attempt to publish it.
...@@ -90,7 +91,7 @@ impl Request {...@@ -90,7 +91,7 @@ impl Request {
90 Self::Conflicts { reply } => reply(Err(error())),91 Self::Conflicts { reply } => reply(Err(error())),
91 Self::Versions { reply } => reply(Err(error())),92 Self::Versions { reply } => reply(Err(error())),
92 Self::Version { reply, .. } => reply(Err(error())),93 Self::Version { reply, .. } => reply(Err(error())),
93 Self::Flush { reply } => reply(Err(error())),94 Self::Flush { reply, .. } => reply(Err(error())),
94 Self::Seal { reply } => reply(Err(error())),95 Self::Seal { reply } => reply(Err(error())),
95 Self::Rebase { reply, .. } => reply(Err(error())),96 Self::Rebase { reply, .. } => reply(Err(error())),
96 Self::Reopen { reply } => reply(Err(error())),97 Self::Reopen { reply } => reply(Err(error())),
...@@ -109,10 +110,16 @@ pub(crate) struct Thread {...@@ -109,10 +110,16 @@ pub(crate) struct Thread {
109 pub(crate) worker: Mutex<Weak<Signal>>,110 pub(crate) worker: Mutex<Weak<Signal>>,
110 /// Taken when the replica drops, which ends every section thread.111 /// Taken when the replica drops, which ends every section thread.
111 sender: Mutex<Option<mpsc::Sender<Request>>>,112 sender: Mutex<Option<mpsc::Sender<Request>>>,
113 #[cfg(target_arch = "wasm32")]
114 durable: Mutex<Option<mpsc::Sender<Receipt>>>,
115 #[cfg(target_arch = "wasm32")]
116 durable_wake: crate::task::Sender<()>,
112 #[cfg(not(target_arch = "wasm32"))]117 #[cfg(not(target_arch = "wasm32"))]
113 threads: Mutex<Vec<std::thread::JoinHandle<()>>>,118 threads: Mutex<Vec<std::thread::JoinHandle<()>>>,
114 /// A password-protected section's key, under which its queue is sealed too.119 /// A password-protected section's key, under which its queue is sealed too.
115 pub(crate) key: Option<Key>,120 pub(crate) key: Option<Key>,
121 /// Last, so the cache closes before background synchronization takes it over.
122 pub(crate) released: crate::Released,
116}123}
117124
118impl Thread {125impl Thread {
...@@ -127,13 +134,19 @@ impl Thread {...@@ -127,13 +134,19 @@ impl Thread {
127 Ok(())134 Ok(())
128 }135 }
129136
130 /// Ends the section threads and waits for them, so the cache is released.137 /// Ends request serving; pending browser receipts retain the cache until they settle.
131 pub(crate) fn stop(&self) {138 pub(crate) fn stop(&self) {
132 if let Ok(mut sender) = self.sender.lock() {139 if let Ok(mut sender) = self.sender.lock() {
133 sender.take();140 sender.take();
134 }141 }
135 #[cfg(target_arch = "wasm32")]142 #[cfg(target_arch = "wasm32")]
136 serving::serve(self);143 {
144 serving::serve(self);
145 if let Ok(mut durable) = self.durable.lock() {
146 durable.take();
147 }
148 let _ = self.durable_wake.try_send(());
149 }
137 #[cfg(not(target_arch = "wasm32"))]150 #[cfg(not(target_arch = "wasm32"))]
138 while let Some(thread) = self151 while let Some(thread) = self
139 .threads152 .threads
...@@ -169,6 +182,77 @@ impl Thread {...@@ -169,6 +182,77 @@ impl Thread {
169 serving::start(self, Box::pin(run(Arc::clone(self))));182 serving::start(self, Box::pin(run(Arc::clone(self))));
170 Ok(())183 Ok(())
171 }184 }
185
186 #[cfg(target_arch = "wasm32")]
187 fn receipt(self: &Arc<Self>, replies: Vec<Reply<()>>, edited: bool) {
188 let receipt = Receipt {
189 shared: Arc::clone(self),
190 replies,
191 edited,
192 };
193 let sent = match self.durable.lock() {
194 Ok(sender) => match sender.as_ref() {
195 Some(sender) => sender.send(receipt).map_err(|error| error.0),
196 None => Err(receipt),
197 },
198 Err(_) => Err(receipt),
199 };
200 if let Err(receipt) = sent {
201 for reply in receipt.replies {
202 reply(Err(io::Error::other("The section thread stopped").into()));
203 }
204 }
205 let _ = self.durable_wake.try_send(());
206 }
207}
208
209#[cfg(target_arch = "wasm32")]
210struct Receipt {
211 shared: Arc<Thread>,
212 replies: Vec<Reply<()>>,
213 edited: bool,
214}
215
216#[cfg(target_arch = "wasm32")]
217async fn receipts(requests: mpsc::Receiver<Receipt>, wake: crate::task::Receiver<()>) {
218 loop {
219 let Receipt {
220 shared,
221 replies,
222 edited,
223 } = match requests.try_recv() {
224 Ok(receipt) => receipt,
225 Err(mpsc::TryRecvError::Empty) => {
226 crate::task::wait(&wake, None).await;
227 continue;
228 }
229 Err(mpsc::TryRecvError::Disconnected) => return,
230 };
231 let result = loop {
232 match crate::fs::durable().await {
233 Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
234 crate::task::wait(&wake, Some(std::time::Duration::from_secs(1))).await;
235 }
236 result => break result,
237 }
238 };
239 match result {
240 Ok(()) => {
241 for reply in replies {
242 reply(Ok(()));
243 }
244 if edited {
245 crate::edited(&shared.worker);
246 }
247 }
248 Err(error) => {
249 let message = error.to_string();
250 for reply in replies {
251 reply(Err(io::Error::new(error.kind(), message.clone()).into()));
252 }
253 }
254 }
255 }
172}256}
173257
174/// The browser's section threads: each replica's `run`, polled whenever a request is sent258/// 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<Request>) -> Option<Request> {...@@ -225,14 +309,25 @@ async fn receive(requests: &mpsc::Receiver<Request>) -> Option<Request> {
225/// Starts the section thread once the queue opens.309/// Starts the section thread once the queue opens.
226pub(crate) fn spawn(connection: Connection, key: Option<Key>) -> Result<(Arc<Thread>, ExGuid)> {310pub(crate) fn spawn(connection: Connection, key: Option<Key>) -> Result<(Arc<Thread>, ExGuid)> {
227 let (sender, requests) = mpsc::channel();311 let (sender, requests) = mpsc::channel();
312 #[cfg(target_arch = "wasm32")]
313 let (durable, written) = mpsc::channel();
314 #[cfg(target_arch = "wasm32")]
315 let (durable_wake, woken) = crate::task::channel();
228 let shared = Arc::new(Thread {316 let shared = Arc::new(Thread {
229 connection: Mutex::new(connection),317 connection: Mutex::new(connection),
230 worker: Mutex::new(Weak::new()),318 worker: Mutex::new(Weak::new()),
231 sender: Mutex::new(Some(sender)),319 sender: Mutex::new(Some(sender)),
320 #[cfg(target_arch = "wasm32")]
321 durable: Mutex::new(Some(durable)),
322 #[cfg(target_arch = "wasm32")]
323 durable_wake,
232 #[cfg(not(target_arch = "wasm32"))]324 #[cfg(not(target_arch = "wasm32"))]
233 threads: Mutex::new(Vec::new()),325 threads: Mutex::new(Vec::new()),
234 key,326 key,
327 released: crate::Released::default(),
235 });328 });
329 #[cfg(target_arch = "wasm32")]
330 crate::task::spawn("onestore-durable", move || receipts(written, woken))?;
236 let (ready, opened) = mpsc::sync_channel(1);331 let (ready, opened) = mpsc::sync_channel(1);
237 shared.start(move |shared| run(shared, requests, VecDeque::new(), Some(ready)))?;332 shared.start(move |shared| run(shared, requests, VecDeque::new(), Some(ready)))?;
238 match opened.recv() {333 match opened.recv() {
...@@ -486,13 +581,25 @@ impl<'a> Working<'a> {...@@ -486,13 +581,25 @@ impl<'a> Working<'a> {
486 later.push_back(request);581 later.push_back(request);
487 false582 false
488 }583 }
489 Request::Flush { reply } => {584 Request::Flush { reply, durable } => {
490 let written = self.flush(shared, &mut accepted);585 let written = self.flush(shared, &mut accepted);
491 reply(if written {586 if written {
492 Ok(())587 #[cfg(target_arch = "wasm32")]
588 if durable {
589 shared.receipt(vec![reply], false);
590 } else {
591 reply(Ok(()));
592 }
593 #[cfg(not(target_arch = "wasm32"))]
594 {
595 let _ = durable;
596 reply(Ok(()));
597 }
493 } else {598 } else {
494 Err(io::Error::other("The queue could not be written").into())599 reply(Err(
495 });600 io::Error::other("The queue could not be written").into()
601 ));
602 }
496 !written603 !written
497 }604 }
498 Request::Seal { reply } => {605 Request::Seal { reply } => {
...@@ -586,7 +693,7 @@ impl<'a> Working<'a> {...@@ -586,7 +693,7 @@ impl<'a> Working<'a> {
586 edit: Edit,693 edit: Edit,
587 reply: Reply<u64>,694 reply: Reply<u64>,
588 accepted: &mut Vec<Accepted>,695 accepted: &mut Vec<Accepted>,
589 shared: &Thread,696 shared: &Arc<Thread>,
590 ) -> std::result::Result<(), bool> {697 ) -> std::result::Result<(), bool> {
591 match self.section.apply(&author, &edit) {698 match self.section.apply(&author, &edit) {
592 Ok(()) => {699 Ok(()) => {
...@@ -612,7 +719,7 @@ impl<'a> Working<'a> {...@@ -612,7 +719,7 @@ impl<'a> Working<'a> {
612719
613 /// Writes the accepted edits in one transaction, then answers their senders; false720 /// Writes the accepted edits in one transaction, then answers their senders; false
614 /// when the write failed and the section holds edits the queue lacks.721 /// when the write failed and the section holds edits the queue lacks.
615 fn flush(&mut self, shared: &Thread, accepted: &mut Vec<Accepted>) -> bool {722 fn flush(&mut self, shared: &Arc<Thread>, accepted: &mut Vec<Accepted>) -> bool {
616 if accepted.is_empty() {723 if accepted.is_empty() {
617 return true;724 return true;
618 }725 }
...@@ -645,10 +752,26 @@ impl<'a> Working<'a> {...@@ -645,10 +752,26 @@ impl<'a> Working<'a> {
645 let ok = written.is_ok();752 let ok = written.is_ok();
646 match written {753 match written {
647 Ok(ids) => {754 Ok(ids) => {
648 for (edit, id) in accepted.drain(..).zip(ids) {755 #[cfg(target_arch = "wasm32")]
649 (edit.reply)(Ok(id));756 shared.receipt(
757 accepted
758 .drain(..)
759 .zip(ids)
760 .map(|(edit, id)| {
761 let reply: Reply<()> =
762 Box::new(move |result| (edit.reply)(result.map(|()| id)));
763 reply
764 })
765 .collect(),
766 true,
767 );
768 #[cfg(not(target_arch = "wasm32"))]
769 {
770 for (edit, id) in accepted.drain(..).zip(ids) {
771 (edit.reply)(Ok(id));
772 }
773 crate::edited(&shared.worker);
650 }774 }
651 crate::edited(&shared.worker);
652 }775 }
653 Err(error) => {776 Err(error) => {
654 let message = error.to_string();777 let message = error.to_string();
crates/notebook/tests/cache.rs+35
...@@ -132,6 +132,41 @@ fn cache_reopen_preserves_the_base_and_queued_edits() {...@@ -132,6 +132,41 @@ fn cache_reopen_preserves_the_base_and_queued_edits() {
132 assert_eq!(replica.pending().unwrap().len(), 2);132 assert_eq!(replica.pending().unwrap().len(), 2);
133}133}
134134
135#[test]
136fn async_edit_receipts_reopen_with_the_same_id_and_text() {
137 use std::{
138 future::Future,
139 task::{Context, Poll, Waker},
140 };
141 let dir = tempfile::tempdir().unwrap();
142 let path = dir.path().join("async.sqlite");
143 let source = onestore::create_section("Async.one", "Receipt", "Fixture").unwrap();
144 let (space, text, _) = target(&source);
145 let replica = Replica::create(&path, &source).unwrap();
146 let id = {
147 let mut receipt =
148 std::pin::pin!(replica.apply_async("Author", typed(space, text, "Durable ")));
149 match receipt
150 .as_mut()
151 .poll(&mut Context::from_waker(Waker::noop()))
152 {
153 Poll::Ready(result) => result.unwrap(),
154 Poll::Pending => panic!("Native durability must finish without a pending wait"),
155 }
156 };
157 {
158 let mut flush = std::pin::pin!(replica.written_async());
159 assert!(matches!(
160 flush.as_mut().poll(&mut Context::from_waker(Waker::noop())),
161 Poll::Ready(Ok(()))
162 ));
163 }
164 drop(replica);
165 let reopened = Replica::open(&path).unwrap();
166 assert_eq!(reopened.pending().unwrap()[0].id, id);
167 assert_eq!(target(&reopened.snapshot().unwrap()).2, "Durable Receipt");
168}
169
135#[test]170#[test]
136fn refused_edits_and_failed_writes_leave_the_queue_as_it_was() {171fn refused_edits_and_failed_writes_leave_the_queue_as_it_was() {
137 let dir = tempfile::tempdir().unwrap();172 let dir = tempfile::tempdir().unwrap();
crates/snowbound/src/history.rs+75-50
...@@ -169,13 +169,17 @@ impl State {...@@ -169,13 +169,17 @@ impl State {
169 version: ExGuid,169 version: ExGuid,
170 ) -> Result<(), Box<dyn Error>> {170 ) -> Result<(), Box<dyn Error>> {
171 let session = self.session.as_mut().ok_or("No section is open")?;171 let session = self.session.as_mut().ok_or("No section is open")?;
172 session172 let operation = session
173 .section173 .section
174 .restore_version(page, version, &self.author)?;174 .restore_version_async(page, version, &self.author);
175 session.refresh_history()?;175 self.section_edit(operation, move |state, _| {
176 self.edited(vec![page]);176 if let Some(session) = &mut state.session {
177 self.commands.push(Command::OpenPage(page));177 session.refresh_history()?;
178 Ok(())178 }
179 state.edited(vec![page]);
180 state.commands.push(Command::OpenPage(page));
181 Ok(())
182 })
179 }183 }
180184
181 pub(crate) fn delete_version(185 pub(crate) fn delete_version(
...@@ -184,10 +188,16 @@ impl State {...@@ -184,10 +188,16 @@ impl State {
184 version: ExGuid,188 version: ExGuid,
185 ) -> Result<(), Box<dyn Error>> {189 ) -> Result<(), Box<dyn Error>> {
186 let session = self.session.as_mut().ok_or("No section is open")?;190 let session = self.session.as_mut().ok_or("No section is open")?;
187 session.section.delete_versions(&[(page, vec![version])])?;191 let operation = session
188 session.refresh_history()?;192 .section
189 self.commands.push(Command::OpenPage(page));193 .delete_versions_async(&[(page, vec![version])]);
190 Ok(())194 self.section_edit(operation, move |state, _| {
195 if let Some(session) = &mut state.session {
196 session.refresh_history()?;
197 }
198 state.commands.push(Command::OpenPage(page));
199 Ok(())
200 })
191 }201 }
192202
193 /// Delete All Versions in the section, its group or the notebook, once confirmed.203 /// Delete All Versions in the section, its group or the notebook, once confirmed.
...@@ -232,55 +242,70 @@ impl State {...@@ -232,55 +242,70 @@ impl State {
232 }) else {242 }) else {
233 return Ok(());243 return Ok(());
234 };244 };
235 clear(&session.section)?;245 let operation = clear(&session.section);
236 session.refresh_history()?;
237 if session.version.is_some() {
238 self.commands.push(Command::OpenPage(session.space));
239 }
240 if scope == Scope::Section {
241 return Ok(());
242 }
243 // The other sections, each opened, cleared and closed in the background.
244 let folder = match scope {
245 Scope::Group => folder,
246 _ => String::new(),
247 };
248 let library = Arc::clone(&session.library);246 let library = Arc::clone(&session.library);
249 let proxy = self.proxy.clone();247 let proxy = self.proxy.clone();
250 crate::spawn(move || {248 let operation = async move {
251 for path in sections(&library, &folder) {249 operation.await?;
252 if path == open {250 if scope != Scope::Section {
253 continue;251 let folder = if scope == Scope::Group {
254 }252 folder
255 let cleared = library253 } else {
256 .open(&path, notify(proxy.clone()))254 String::new()
257 .and_then(|section| {255 };
258 clear(&section)?;256 crate::spawn_async(move || async move {
259 Ok(section.close()?)257 for path in sections(&library, &folder) {
260 });258 if path == open {
261 if let Err(error) = cleared {259 continue;
262 eprintln!("Deleting the versions in {path} failed: {error}");260 }
261 let cleared: Result<(), Box<dyn Error>> = async {
262 let section = library.open(&path, notify(proxy.clone()))?;
263 clear(&section).await?;
264 Ok(section.close_async().await?)
265 }
266 .await;
267 if let Err(error) = cleared {
268 eprintln!("Deleting the versions in {path} failed: {error}");
269 }
270 }
271 });
272 }
273 Ok(())
274 };
275 self.section_edit(operation, move |state, ()| {
276 if let Some(session) = &mut state.session {
277 session.refresh_history()?;
278 if session.version.is_some() {
279 state.commands.push(Command::OpenPage(session.space));
263 }280 }
264 }281 }
265 });282 Ok(())
266 Ok(())283 })
267 }284 }
268}285}
269286
270/// Deletes every page version `section` holds.287/// Deletes every page version `section` holds.
271fn clear(section: &notebook::session::Section) -> Result<(), notebook::Error> {288fn clear(
272 let all: Vec<(ExGuid, Vec<ExGuid>)> = (section.versions()?.into_iter())289 section: &notebook::session::Section,
273 .map(|(page, versions)| {290) -> impl std::future::Future<Output = Result<(), notebook::Error>> + Send + 'static + use<> {
274 (291 let operation = section.versions().map(|versions| {
275 page,292 let all: Vec<(ExGuid, Vec<ExGuid>)> = versions
276 versions.iter().map(|version| version.context).collect(),293 .into_iter()
277 )294 .map(|(page, versions)| {
278 })295 (
279 .collect();296 page,
280 if !all.is_empty() {297 versions.iter().map(|version| version.context).collect(),
281 section.delete_versions(&all)?;298 )
299 })
300 .collect();
301 (!all.is_empty()).then(|| section.delete_versions_async(&all))
302 });
303 async move {
304 if let Some(operation) = operation? {
305 operation.await?;
306 }
307 Ok(())
282 }308 }
283 Ok(())
284}309}
285310
286/// The readable sections at catalog path `folder` and in the groups inside it, leaving out311/// The readable sections at catalog path `folder` and in the groups inside it, leaving out
crates/snowbound/src/library.rs+47-1
...@@ -974,6 +974,41 @@ impl Library {...@@ -974,6 +974,41 @@ impl Library {
974 }974 }
975 }975 }
976976
977 pub async fn close_kept_async(&self) -> Result<(), Box<dyn Error>> {
978 #[cfg(not(target_arch = "wasm32"))]
979 {
980 self.close_kept();
981 Ok(())
982 }
983 #[cfg(target_arch = "wasm32")]
984 {
985 let replicas: Vec<_> = {
986 let mut kept = self
987 .kept
988 .lock()
989 .map_err(|_| io::Error::from(io::ErrorKind::Other))?;
990 kept.values_mut()
991 .map(|section| Arc::clone(section.replica()))
992 .collect()
993 };
994 for replica in replicas {
995 replica.written_async().await?;
996 }
997 let stopping: Vec<_> = {
998 let mut kept = self
999 .kept
1000 .lock()
1001 .map_err(|_| io::Error::from(io::ErrorKind::Other))?;
1002 kept.values_mut().map(Section::stop_async).collect()
1003 };
1004 for stopping in stopping {
1005 stopping.await?;
1006 }
1007 self.close_kept();
1008 Ok(())
1009 }
1010 }
1011
977 /// Where `file`, a section file of this notebook, sits on this computer: under the1012 /// Where `file`, a section file of this notebook, sits on this computer: under the
978 /// mount it was opened from when the embedded client holds it share-relative.1013 /// mount it was opened from when the embedded client holds it share-relative.
979 pub fn local(&self, file: &Path) -> Option<PathBuf> {1014 pub fn local(&self, file: &Path) -> Option<PathBuf> {
...@@ -1186,10 +1221,21 @@ impl Library {...@@ -1186,10 +1221,21 @@ impl Library {
1186 /// Unlocks the section at catalog `path` with `password`, as OneNote's Protected Section1221 /// Unlocks the section at catalog `path` with `password`, as OneNote's Protected Section
1187 /// dialog does; a wrong password is `PasswordMismatch`.1222 /// dialog does; a wrong password is `PasswordMismatch`.
1188 pub fn unlock(&self, path: &str, password: &str) -> Result<(), notebook::Error> {1223 pub fn unlock(&self, path: &str, password: &str) -> Result<(), notebook::Error> {
1224 let operation = self.unlock_async(path, password);
1225 #[cfg(not(target_arch = "wasm32"))]
1226 return pollster::block_on(operation);
1227 #[cfg(target_arch = "wasm32")]
1228 {
1229 drop(operation);
1230 Err(io::ErrorKind::WouldBlock.into())
1231 }
1232 }
1233
1234 pub async fn unlock_async(&self, path: &str, password: &str) -> Result<(), notebook::Error> {
1189 let Ok(Some(notebook)) = &self.notebook else {1235 let Ok(Some(notebook)) = &self.notebook else {
1190 return Err(io::Error::from(io::ErrorKind::NotFound).into());1236 return Err(io::Error::from(io::ErrorKind::NotFound).into());
1191 };1237 };
1192 let key = notebook.unlock(path, password)?;1238 let key = notebook.unlock_async(path, password).await?;
1193 self.keep_key(path, key);1239 self.keep_key(path, key);
1194 Ok(())1240 Ok(())
1195 }1241 }
crates/snowbound/src/main.rs+70-14
...@@ -3582,9 +3582,14 @@ impl State {...@@ -3582,9 +3582,14 @@ impl State {
3582 }3582 }
3583 Command::DeleteVersion { page, version } => {3583 Command::DeleteVersion { page, version } => {
3584 let session = self.session.as_mut().ok_or("No section is open")?;3584 let session = self.session.as_mut().ok_or("No section is open")?;
3585 session.section.delete_pages(&[version])?;3585 let operation = session.section.delete_pages_async(&[version]);
3586 session.refresh_conflicts()?;3586 self.section_edit(operation, move |state, _| {
3587 self.commands.push(Command::OpenPage(page));3587 if let Some(session) = &mut state.session {
3588 session.refresh_conflicts()?;
3589 }
3590 state.commands.push(Command::OpenPage(page));
3591 Ok(())
3592 })?;
3588 }3593 }
3589 Command::CopyVersion { version, tab } => {3594 Command::CopyVersion { version, tab } => {
3590 let session = self.session.as_mut().ok_or("No section is open")?;3595 let session = self.session.as_mut().ok_or("No section is open")?;
...@@ -3595,18 +3600,25 @@ impl State {...@@ -3595,18 +3600,25 @@ impl State {
3595 _ => session.section.page(version)?,3600 _ => session.section.page(version)?,
3596 };3601 };
3597 if tab == session.tab {3602 if tab == session.tab {
3598 session.section.import_page(&page, &self.author)?;3603 let operation = session.section.import_page_async(&page, &self.author);
3599 session.pages = session.section.pages()?;3604 self.section_edit(operation, |state, _| {
3600 self.edited(Vec::new());3605 if let Some(session) = &mut state.session {
3606 session.pages = session.section.pages()?;
3607 }
3608 state.edited(Vec::new());
3609 Ok(())
3610 })?;
3601 } else {3611 } else {
3602 let library = Arc::clone(&session.library);3612 let library = Arc::clone(&session.library);
3603 let path = session.tabs[tab].path.clone();3613 let path = session.tabs[tab].path.clone();
3604 let (notify, author) = (notify(self.proxy.clone()), self.author.clone());3614 let (notify, author) = (notify(self.proxy.clone()), self.author.clone());
3605 crate::spawn(move || {3615 crate::spawn_async(move || async move {
3606 let copied = library.open(&path, notify).and_then(|section| {3616 let copied: Result<(), Box<dyn Error>> = async {
3607 section.import_page(&page, &author)?;3617 let section = library.open(&path, notify)?;
3608 Ok(section.close()?)3618 section.import_page_async(&page, &author).await?;
3609 });3619 Ok(section.close_async().await?)
3620 }
3621 .await;
3610 if let Err(error) = copied {3622 if let Err(error) = copied {
3611 eprintln!("Copying the page failed: {error}");3623 eprintln!("Copying the page failed: {error}");
3612 }3624 }
...@@ -3676,14 +3688,21 @@ impl State {...@@ -3676,14 +3688,21 @@ impl State {
3676 /// Runs `read` on a thread of its own, which lays out the page it read; `open_loaded`3688 /// Runs `read` on a thread of its own, which lays out the page it read; `open_loaded`
3677 /// shows it unless a newer read was asked for meanwhile.3689 /// shows it unless a newer read was asked for meanwhile.
3678 fn load(&mut self, read: impl FnOnce() -> Result<Loaded, Box<dyn Error>> + Send + 'static) {3690 fn load(&mut self, read: impl FnOnce() -> Result<Loaded, Box<dyn Error>> + Send + 'static) {
3691 self.load_async(move || async move { read() });
3692 }
3693
3694 fn load_async<F: Future<Output = Result<Loaded, Box<dyn Error>>> + 'static>(
3695 &mut self,
3696 read: impl FnOnce() -> F + Send + 'static,
3697 ) {
3679 self.loading += 1;3698 self.loading += 1;
3680 let (id, sender, redraw) = (self.loading, self.loads.0.clone(), self.redraw.clone());3699 let (id, sender, redraw) = (self.loading, self.loads.0.clone(), self.redraw.clone());
3681 let layouts = Arc::clone(&self.layouts);3700 let layouts = Arc::clone(&self.layouts);
3682 let requested = Instant::now();3701 let requested = Instant::now();
3683 let scenes = Arc::clone(&self.prefetch.scenes);3702 let scenes = Arc::clone(&self.prefetch.scenes);
3684 let open = self.session.as_ref().map(Session::key);3703 let open = self.session.as_ref().map(Session::key);
3685 crate::spawn(move || {3704 crate::spawn_async(move || async move {
3686 let laid = read().and_then(|loaded| {3705 let laid = read().await.and_then(|loaded| {
3687 lap("switch read", requested);3706 lap("switch read", requested);
3688 let (shown, page) = match loaded {3707 let (shown, page) = match loaded {
3689 Loaded::Section(session, page) => (Shown::Section(session), page),3708 Loaded::Section(session, page) => (Shown::Section(session), page),
...@@ -3730,6 +3749,44 @@ impl State {...@@ -3730,6 +3749,44 @@ impl State {
3730 });3749 });
3731 }3750 }
37323751
3752 fn complete_async<T: Send + 'static, F: Future<Output = T> + Send + 'static>(
3753 &mut self,
3754 operation: F,
3755 then: impl FnOnce(&mut State, T) -> Result<(), Box<dyn Error>> + Send + 'static,
3756 ) -> Result<(), Box<dyn Error>> {
3757 #[cfg(not(target_arch = "wasm32"))]
3758 return then(self, pollster::block_on(operation));
3759 #[cfg(target_arch = "wasm32")]
3760 {
3761 let reply = self.reply(then);
3762 crate::spawn_async(move || async move { reply.send(operation.await) });
3763 Ok(())
3764 }
3765 }
3766
3767 fn section_edit<
3768 T: Send + 'static,
3769 F: Future<Output = Result<T, notebook::Error>> + Send + 'static,
3770 >(
3771 &mut self,
3772 operation: F,
3773 then: impl FnOnce(&mut State, T) -> Result<(), Box<dyn Error>> + Send + 'static,
3774 ) -> Result<(), Box<dyn Error>> {
3775 let key = self.session.as_ref().ok_or("No section is open")?.key();
3776 self.complete_async(operation, move |state, result| {
3777 let value = result?;
3778 if state
3779 .session
3780 .as_ref()
3781 .is_some_and(|session| session.key() == key)
3782 {
3783 then(state, value)
3784 } else {
3785 Ok(())
3786 }
3787 })
3788 }
3789
3733 /// Shows the newest page read once the pictures it shows first are drawn, or after3790 /// Shows the newest page read once the pictures it shows first are drawn, or after
3734 /// `HOLD`.3791 /// `HOLD`.
3735 fn open_loaded(&mut self) -> Result<(), Box<dyn Error>> {3792 fn open_loaded(&mut self) -> Result<(), Box<dyn Error>> {
...@@ -6207,7 +6264,6 @@ fn spawn(work: impl FnOnce() + Send + 'static) {...@@ -6207,7 +6264,6 @@ fn spawn(work: impl FnOnce() + Send + 'static) {
6207}6264}
62086265
6209/// Runs asynchronous work beside the frame, creating the future on its executor.6266/// Runs asynchronous work beside the frame, creating the future on its executor.
6210#[cfg(feature = "live")]
6211fn spawn_async<F: Future<Output = ()> + 'static>(work: impl FnOnce() -> F + Send + 'static) {6267fn spawn_async<F: Future<Output = ()> + 'static>(work: impl FnOnce() -> F + Send + 'static) {
6212 RUNNING.fetch_add(1, Ordering::Relaxed);6268 RUNNING.fetch_add(1, Ordering::Relaxed);
6213 #[cfg(not(target_arch = "wasm32"))]6269 #[cfg(not(target_arch = "wasm32"))]
crates/snowbound/src/manage.rs+2-2
...@@ -781,9 +781,9 @@ impl State {...@@ -781,9 +781,9 @@ impl State {
781 .filter(|session| Arc::ptr_eq(&session.library, &library))781 .filter(|session| Arc::ptr_eq(&session.library, &library))
782 .map(|session| session.tabs[session.tab].path.clone());782 .map(|session| session.tabs[session.tab].path.clone());
783 let notify = notify(self.proxy.clone());783 let notify = notify(self.proxy.clone());
784 self.load(move || {784 self.load_async(move || async move {
785 // A section kept open would hold a file the change moves.785 // A section kept open would hold a file the change moves.
786 library.close_kept();786 library.close_kept_async().await?;
787 let mut notebook = library.reopen()?;787 let mut notebook = library.reopen()?;
788 let names = |notebook: &Notebook, folder: &str| -> Vec<String> {788 let names = |notebook: &Notebook, folder: &str| -> Vec<String> {
789 let found = crate::library::folders(notebook.catalog(), |_| true)789 let found = crate::library::folders(notebook.catalog(), |_| true)
crates/snowbound/src/protection.rs+192-67
...@@ -565,23 +565,9 @@ impl State {...@@ -565,23 +565,9 @@ impl State {
565 (Some("Set Password"), _) => Some(Dialog::Set(Default::default(), None)),565 (Some("Set Password"), _) => Some(Dialog::Set(Default::default(), None)),
566 (Some("Change Password"), _) => Some(Dialog::Change(Default::default(), None)),566 (Some("Change Password"), _) => Some(Dialog::Change(Default::default(), None)),
567 (Some("Remove Password"), _) => Some(Dialog::Remove(Zeroizing::default(), false)),567 (Some("Remove Password"), _) => Some(Dialog::Remove(Zeroizing::default(), false)),
568 (Some(_), Dialog::Unlock(value, _)) => match library.unlock(&path, &value) {568 (Some(_), dialog @ Dialog::Unlock(..)) => {
569 Ok(()) => {569 return self.submit_password(library, path, dialog);
570 self.commands570 }
571 .push(Command::OpenSection(Arc::clone(&library), path.clone()));
572 None
573 }
574 Err(notebook::Error::Protected(_)) => {
575 Some(Dialog::Unlock(Zeroizing::default(), true))
576 }
577 Err(error) => {
578 crate::platform::alert(
579 "Couldn't unlock the section",
580 &crate::plain(&error, "section"),
581 );
582 None
583 }
584 },
585 (Some(_), Dialog::Set([value, confirm], _)) => {571 (Some(_), Dialog::Set([value, confirm], _)) => {
586 if value.is_empty() {572 if value.is_empty() {
587 Some(Dialog::Set(Default::default(), Some("Enter a password.")))573 Some(Dialog::Set(Default::default(), Some("Enter a password.")))
...@@ -607,30 +593,16 @@ impl State {...@@ -607,30 +593,16 @@ impl State {
607 Some("The new passwords don\u{2019}t match."),593 Some("The new passwords don\u{2019}t match."),
608 ))594 ))
609 } else {595 } else {
610 match current(&library, &path, &old) {596 return self.submit_password(
611 Some(key) => {597 library,
612 self.set_password(598 path,
613 library.clone(),599 Dialog::Change([old, value, confirm], None),
614 path.clone(),600 );
615 Some(key),
616 Some(value),
617 );
618 None
619 }
620 None => Some(Dialog::Change(
621 Default::default(),
622 Some("Password is incorrect."),
623 )),
624 }
625 }601 }
626 }602 }
627 (Some(_), Dialog::Remove(value, _)) => match current(&library, &path, &value) {603 (Some(_), dialog @ Dialog::Remove(..)) => {
628 Some(key) => {604 return self.submit_password(library, path, dialog);
629 self.set_password(library.clone(), path.clone(), Some(key), None);605 }
630 None
631 }
632 None => Some(Dialog::Remove(Zeroizing::default(), true)),
633 },
634 (Some(_), Dialog::Protection) => None,606 (Some(_), Dialog::Protection) => None,
635 };607 };
636 match next {608 match next {
...@@ -640,6 +612,84 @@ impl State {...@@ -640,6 +612,84 @@ impl State {
640 Ok(())612 Ok(())
641 }613 }
642614
615 fn submit_password(
616 &mut self,
617 library: Arc<Library>,
618 path: String,
619 dialog: Dialog,
620 ) -> Result<(), Box<dyn Error>> {
621 #[cfg(target_arch = "wasm32")]
622 self.ui.close_popup(id());
623 let operation = async move {
624 let result = {
625 let password = match &dialog {
626 Dialog::Unlock(value, _) | Dialog::Remove(value, _) => value,
627 Dialog::Change([old, ..], _) => old,
628 _ => unreachable!("Only password checks are submitted"),
629 };
630 #[cfg(not(target_arch = "wasm32"))]
631 let result = library.unlock(&path, password);
632 #[cfg(target_arch = "wasm32")]
633 let result = library.unlock_async(&path, password).await;
634 result
635 };
636 (library, path, dialog, result)
637 };
638 self.complete_async(operation, |state, (library, path, dialog, result)| {
639 if let Err(error) = &result
640 && !matches!(error, notebook::Error::Protected(_))
641 {
642 state.ui.close_popup(id());
643 crate::platform::alert(
644 "Couldn't unlock the section",
645 &crate::plain(error, "section"),
646 );
647 return Ok(());
648 }
649 let key = result.as_ref().ok().and_then(|()| library.unlocked(&path));
650 let next = match dialog {
651 Dialog::Unlock(..) => {
652 if result.is_ok() {
653 state
654 .commands
655 .push(Command::OpenSection(Arc::clone(&library), path.clone()));
656 None
657 } else {
658 Some(Dialog::Unlock(Zeroizing::default(), true))
659 }
660 }
661 Dialog::Change([_, value, _], _) => match key {
662 Some(key) => {
663 state.set_password(
664 Arc::clone(&library),
665 path.clone(),
666 Some(key),
667 Some(value),
668 );
669 None
670 }
671 None => Some(Dialog::Change(
672 Default::default(),
673 Some("Password is incorrect."),
674 )),
675 },
676 Dialog::Remove(..) => match key {
677 Some(key) => {
678 state.set_password(Arc::clone(&library), path.clone(), Some(key), None);
679 None
680 }
681 None => Some(Dialog::Remove(Zeroizing::default(), true)),
682 },
683 _ => unreachable!("Only password checks are submitted"),
684 };
685 match next {
686 Some(dialog) => state.ask_password(library, path, dialog),
687 None => state.ui.close_popup(id()),
688 }
689 Ok(())
690 })
691 }
692
643 /// Sets, changes or removes the password of `path` in `library`, closing its section693 /// Sets, changes or removes the password of `path` in `library`, closing its section
644 /// first: the section is written anew, and its replica goes with the old file.694 /// first: the section is written anew, and its replica goes with the old file.
645 fn set_password(695 fn set_password(
...@@ -649,38 +699,113 @@ impl State {...@@ -649,38 +699,113 @@ impl State {
649 key: Option<onestore::protected::Key>,699 key: Option<onestore::protected::Key>,
650 password: Option<Zeroizing<String>>,700 password: Option<Zeroizing<String>>,
651 ) {701 ) {
652 let open = self.session.as_ref().is_some_and(|session| {702 let replica = self
653 Arc::ptr_eq(&session.library, &library) && session.tabs[session.tab].path == path703 .session
654 });704 .as_ref()
655 if open {705 .filter(|session| {
656 let closed = self.persist().and_then(|()| {706 Arc::ptr_eq(&session.library, &library) && session.tabs[session.tab].path == path
657 if let Some(session) = self.session.take() {707 })
658 session.section.close()?;708 .map(|session| Arc::clone(session.section.replica()));
709 let Some(replica) = replica else {
710 library.lock(|locked, _| locked == path);
711 self.restructure(
712 library,
713 Structure::Password {
714 path,
715 key,
716 password,
717 },
718 );
719 return;
720 };
721 if let Err(error) = self.persist() {
722 return crate::platform::alert(
723 "Couldn't set the password",
724 &crate::plain(&*error, "section"),
725 );
726 }
727 let operation = async move {
728 let through = replica.pending()?.last().map_or(0, |edit| edit.id);
729 replica.written_async().await?;
730 Ok::<_, notebook::Error>((replica, through))
731 };
732 let _ = self.complete_async(operation, move |state, result| {
733 let prepared = (|| -> Result<(), Box<dyn Error>> {
734 let (replica, through) = result?;
735 let open = state
736 .session
737 .as_ref()
738 .is_some_and(|session| Arc::ptr_eq(session.section.replica(), &replica));
739 if open {
740 state.persist()?;
741 let pending = replica.pending()?.last().map_or(0, |edit| edit.id);
742 if pending > through {
743 state.set_password(library, path, key, password);
744 return Ok(());
745 }
659 }746 }
660 Ok(())747 let stopping = state
661 });748 .session
662 if let Err(error) = closed {749 .as_mut()
663 return crate::platform::alert(750 .filter(|_| open)
751 .map(|session| session.section.stop_async());
752 let operation = async move {
753 if let Some(stopping) = stopping {
754 stopping.await?;
755 }
756 let through = replica.pending()?.last().map_or(0, |edit| edit.id);
757 replica.written_async().await?;
758 Ok::<_, notebook::Error>((replica, through))
759 };
760 state.complete_async(operation, move |state, result| {
761 let closed =
762 (|| -> Result<(), Box<dyn Error>> {
763 let (replica, through) = result?;
764 if state.session.as_ref().is_some_and(|session| {
765 Arc::ptr_eq(session.section.replica(), &replica)
766 }) {
767 state.persist()?;
768 if replica
769 .pending()?
770 .last()
771 .is_some_and(|edit| edit.id > through)
772 {
773 state.set_password(library, path, key, password);
774 return Ok(());
775 }
776 if let Some(session) = state.session.take() {
777 session.section.close()?;
778 }
779 }
780 drop(replica);
781 state.prefetch.forget(&library.key(&path));
782 library.lock(|locked, _| locked == path);
783 state.restructure(
784 library,
785 Structure::Password {
786 path,
787 key,
788 password,
789 },
790 );
791 Ok(())
792 })();
793 if let Err(error) = closed {
794 crate::platform::alert(
795 "Couldn't set the password",
796 &crate::plain(&*error, "section"),
797 );
798 }
799 Ok(())
800 })
801 })();
802 if let Err(error) = prepared {
803 crate::platform::alert(
664 "Couldn't set the password",804 "Couldn't set the password",
665 &crate::plain(&*error, "section"),805 &crate::plain(&*error, "section"),
666 );806 );
667 }807 }
668 self.prefetch.forget(&library.key(&path));808 Ok(())
669 }809 });
670 library.lock(|locked, _| locked == path);
671 self.restructure(
672 library,
673 Structure::Password {
674 path,
675 key,
676 password,
677 },
678 );
679 }810 }
680}811}
681
682/// The key `password` opens the section at `path` with, if it does.
683fn current(library: &Library, path: &str, password: &str) -> Option<onestore::protected::Key> {
684 library.unlock(path, password).ok()?;
685 library.unlocked(path)
686}
crates/snowbound/src/recycle.rs+40-23
...@@ -135,8 +135,8 @@ impl State {...@@ -135,8 +135,8 @@ impl State {
135 (spaces.clone().skip(at).find(kept)).or_else(|| spaces.take(at).rev().find(kept))135 (spaces.clone().skip(at).find(kept)).or_else(|| spaces.take(at).rev().find(kept))
136 });136 });
137 let (author, proxy) = (self.author.clone(), self.proxy.clone());137 let (author, proxy) = (self.author.clone(), self.proxy.clone());
138 self.load(move || {138 self.load_async(move || async move {
139 carry_out(&library, &replica, request, &author)?;139 carry_out(&library, &replica, request, &author).await?;
140 if let Some(show) = if gone.contains(&shown) {140 if let Some(show) = if gone.contains(&shown) {
141 after141 after
142 } else {142 } else {
...@@ -192,7 +192,7 @@ impl State {...@@ -192,7 +192,7 @@ impl State {
192/// Does `request` to the bin's section `replica` holds, in `library`, as `author`: a page192/// Does `request` to the bin's section `replica` holds, in `library`, as `author`: a page
193/// restored goes to its section as one op, keeping its identity, a copy under a new one, and193/// restored goes to its section as one op, keeping its identity, a copy under a new one, and
194/// what leaves the bin goes from it as another.194/// what leaves the bin goes from it as another.
195fn carry_out(195async fn carry_out(
196 library: &Library,196 library: &Library,
197 replica: &notebook::Replica,197 replica: &notebook::Replica,
198 request: Request,198 request: Request,
...@@ -203,31 +203,36 @@ fn carry_out(...@@ -203,31 +203,36 @@ fn carry_out(
203 let page = replica.page(space)?;203 let page = replica.page(space)?;
204 let section = library.open(&path, || {})?;204 let section = library.open(&path, || {})?;
205 if copy {205 if copy {
206 section.import_page(&page, author)?;206 section.import_page_async(&page, author).await?;
207 } else {207 } else {
208 let ops = vec![notebook::session::moved(&page, author)?];208 let ops = vec![notebook::session::moved(&page, author)?];
209 section.replica().apply(209 section
210 author,210 .replica()
211 Edit {211 .apply_async(
212 at: filetime(),212 author,
213 ops,213 Edit {
214 },214 at: filetime(),
215 )?;215 ops,
216 },
217 )
218 .await?;
216 }219 }
217 section.close()?;220 section.close_async().await?;
218 (!copy).then_some(vec![space]).unwrap_or_default()221 (!copy).then_some(vec![space]).unwrap_or_default()
219 }222 }
220 Request::Purge(spaces) => spaces,223 Request::Purge(spaces) => spaces,
221 };224 };
222 if !gone.is_empty() {225 if !gone.is_empty() {
223 let ops = vec![Op::Section(SectionOp::Delete(gone))];226 let ops = vec![Op::Section(SectionOp::Delete(gone))];
224 replica.apply(227 replica
225 author,228 .apply_async(
226 Edit {229 author,
227 at: filetime(),230 Edit {
228 ops,231 at: filetime(),
229 },232 ops,
230 )?;233 },
234 )
235 .await?;
231 }236 }
232 Ok(())237 Ok(())
233}238}
...@@ -360,14 +365,26 @@ mod tests {...@@ -360,14 +365,26 @@ mod tests {
360 path: FIRST.into(),365 path: FIRST.into(),
361 copy,366 copy,
362 };367 };
363 carry_out(&library, bin.replica(), restore(back, false), AUTHOR).unwrap();368 pollster::block_on(carry_out(
364 carry_out(&library, bin.replica(), restore(copied, true), AUTHOR).unwrap();369 &library,
365 carry_out(370 bin.replica(),
371 restore(back, false),
372 AUTHOR,
373 ))
374 .unwrap();
375 pollster::block_on(carry_out(
376 &library,
377 bin.replica(),
378 restore(copied, true),
379 AUTHOR,
380 ))
381 .unwrap();
382 pollster::block_on(carry_out(
366 &library,383 &library,
367 bin.replica(),384 bin.replica(),
368 Request::Purge(vec![purged]),385 Request::Purge(vec![purged]),
369 AUTHOR,386 AUTHOR,
370 )387 ))
371 .unwrap();388 .unwrap();
372 bin.close().unwrap();389 bin.close().unwrap();
373 published(&library, FIRST);390 published(&library, FIRST);
crates/snowbound/src/undo.rs+49-35
...@@ -350,14 +350,16 @@ struct Applied {...@@ -350,14 +350,16 @@ struct Applied {
350}350}
351351
352impl Site {352impl Site {
353 fn apply(&self, ops: Vec<Op>) -> Result<(), Box<dyn Error>> {353 async fn apply(&self, ops: Vec<Op>) -> Result<(), Box<dyn Error>> {
354 self.replica.apply(354 self.replica
355 &self.author,355 .apply_async(
356 Edit {356 &self.author,
357 at: crate::filetime(),357 Edit {
358 ops,358 at: crate::filetime(),
359 },359 ops,
360 )?;360 },
361 )
362 .await?;
361 Ok(())363 Ok(())
362 }364 }
363365
...@@ -408,7 +410,11 @@ impl Site {...@@ -408,7 +410,11 @@ impl Site {
408410
409 /// Does `change` to the section's pages, `shown` the page open; `None` where what it411 /// Does `change` to the section's pages, `shown` the page open; `None` where what it
410 /// changes is gone or holds what it did not make, which it leaves alone.412 /// changes is gone or holds what it did not make, which it leaves alone.
411 fn change(&self, change: Change, shown: ExGuid) -> Result<Option<Applied>, Box<dyn Error>> {413 async fn change(
414 &self,
415 change: Change,
416 shown: ExGuid,
417 ) -> Result<Option<Applied>, Box<dyn Error>> {
412 let listed = self.replica.pages()?;418 let listed = self.replica.pages()?;
413 let at = |space: ExGuid| listed.iter().position(|(listed, ..)| *listed == space);419 let at = |space: ExGuid| listed.iter().position(|(listed, ..)| *listed == space);
414 let binned = matches!(self.library.notebook, Ok(Some(_)));420 let binned = matches!(self.library.notebook, Ok(Some(_)));
...@@ -449,8 +455,9 @@ impl Site {...@@ -449,8 +455,9 @@ impl Site {
449 if !created && binned {455 if !created && binned {
450 let pages: Vec<Page> = kept.iter().map(|kept| kept.page.clone()).collect();456 let pages: Vec<Page> = kept.iter().map(|kept| kept.page.clone()).collect();
451 self.library.reopen()?.recycle_pages(&pages, &self.author)?;457 self.library.reopen()?.recycle_pages(&pages, &self.author)?;
458 self.replica.written_async().await?;
452 }459 }
453 self.apply(ops)?;460 self.apply(ops).await?;
454 Applied {461 Applied {
455 undo: Change::Restore {462 undo: Change::Restore {
456 pages: kept,463 pages: kept,
...@@ -494,7 +501,7 @@ impl Site {...@@ -494,7 +501,7 @@ impl Site {
494 spaces.push(space);501 spaces.push(space);
495 }502 }
496 let gone = self.take_added(added, &listed, &mut ops);503 let gone = self.take_added(added, &listed, &mut ops);
497 self.apply(ops)?;504 self.apply(ops).await?;
498 if !created && binned {505 if !created && binned {
499 let identities: Vec<[u8; 16]> =506 let identities: Vec<[u8; 16]> =
500 back.iter().filter_map(|kept| kept.page.identity).collect();507 back.iter().filter_map(|kept| kept.page.identity).collect();
...@@ -515,7 +522,8 @@ impl Site {...@@ -515,7 +522,8 @@ impl Site {
515 let Some(show) = edits.first().map(PageEdit::space) else {522 let Some(show) = edits.first().map(PageEdit::space) else {
516 return Ok(None);523 return Ok(None);
517 };524 };
518 self.apply(vec![Op::Section(SectionOp::Pages(edits))])?;525 self.apply(vec![Op::Section(SectionOp::Pages(edits))])
526 .await?;
519 Applied {527 Applied {
520 undo: Change::Arrange {528 undo: Change::Arrange {
521 order: listed529 order: listed
...@@ -533,7 +541,7 @@ impl Site {...@@ -533,7 +541,7 @@ impl Site {
533 return Ok(None);541 return Ok(None);
534 }542 }
535 let op = crate::rename::retitled(&self.replica.page(page)?, page, to.clone())?;543 let op = crate::rename::retitled(&self.replica.page(page)?, page, to.clone())?;
536 self.apply(vec![op])?;544 self.apply(vec![op]).await?;
537 Applied {545 Applied {
538 undo: Change::Retitle {546 undo: Change::Retitle {
539 page,547 page,
...@@ -555,15 +563,18 @@ impl Site {...@@ -555,15 +563,18 @@ impl Site {
555 let mut ops = vec![Op::Section(SectionOp::Delete(vec![page]))];563 let mut ops = vec![Op::Section(SectionOp::Delete(vec![page]))];
556 let (show, added) = self.after(&listed, &[page], None, &mut ops);564 let (show, added) = self.after(&listed, &[page], None, &mut ops);
557 let section = self.library.open(&path, || {})?;565 let section = self.library.open(&path, || {})?;
558 section.replica().apply(566 section
559 &self.author,567 .replica()
560 Edit {568 .apply_async(
561 at: crate::filetime(),569 &self.author,
562 ops: vec![import],570 Edit {
563 },571 at: crate::filetime(),
564 )?;572 ops: vec![import],
565 section.close()?;573 },
566 self.apply(ops)?;574 )
575 .await?;
576 section.close_async().await?;
577 self.apply(ops).await?;
567 Applied {578 Applied {
568 undo: Change::Return {579 undo: Change::Return {
569 page: arrived,580 page: arrived,
...@@ -592,7 +603,7 @@ impl Site {...@@ -592,7 +603,7 @@ impl Site {
592 false => None,603 false => None,
593 };604 };
594 let Some(sent) = sent else {605 let Some(sent) = sent else {
595 section.close()?;606 section.close_async().await?;
596 return Ok(None);607 return Ok(None);
597 };608 };
598 let import = notebook::session::moved(&sent, &self.author)?;609 let import = notebook::session::moved(&sent, &self.author)?;
...@@ -606,15 +617,18 @@ impl Site {...@@ -606,15 +617,18 @@ impl Site {
606 )?])),617 )?])),
607 ];618 ];
608 let gone = self.take_added(added, &listed, &mut ops);619 let gone = self.take_added(added, &listed, &mut ops);
609 self.apply(ops)?;620 self.apply(ops).await?;
610 section.replica().apply(621 section
611 &self.author,622 .replica()
612 Edit {623 .apply_async(
613 at: crate::filetime(),624 &self.author,
614 ops: vec![Op::Section(SectionOp::Delete(vec![page]))],625 Edit {
615 },626 at: crate::filetime(),
616 )?;627 ops: vec![Op::Section(SectionOp::Delete(vec![page]))],
617 section.close()?;628 },
629 )
630 .await?;
631 section.close_async().await?;
618 Applied {632 Applied {
619 undo: Change::Send { page: back, target },633 undo: Change::Send { page: back, target },
620 show: back,634 show: back,
...@@ -1052,8 +1066,8 @@ impl State {...@@ -1052,8 +1066,8 @@ impl State {
1052 let proxy = self.proxy.clone();1066 let proxy = self.proxy.clone();
1053 self.undo.busy += 1;1067 self.undo.busy += 1;
1054 self.switching = Some((None, web_time::Instant::now()));1068 self.switching = Some((None, web_time::Instant::now()));
1055 self.load(move || {1069 self.load_async(move || async move {
1056 let applied = site.change(change, shown);1070 let applied = site.change(change, shown).await;
1057 let show = match &applied {1071 let show = match &applied {
1058 Ok(Some(applied)) => applied.show,1072 Ok(Some(applied)) => applied.show,
1059 _ => shown,1073 _ => shown,
...@@ -1382,7 +1396,7 @@ mod tests {...@@ -1382,7 +1396,7 @@ mod tests {
1382 /// Does `change`, as the app's thread does, returning what takes it back.1396 /// Does `change`, as the app's thread does, returning what takes it back.
1383 fn change(&self, change: Change) -> Option<Applied> {1397 fn change(&self, change: Change) -> Option<Applied> {
1384 let shown = self.pages()[0].0;1398 let shown = self.pages()[0].0;
1385 self.site().change(change, shown).unwrap()1399 pollster::block_on(self.site().change(change, shown)).unwrap()
1386 }1400 }
1387 }1401 }
13881402
crates/snowbound/src/web.rs+5-4
...@@ -1054,7 +1054,8 @@ fn restore(files: js_sys::Array) {...@@ -1054,7 +1054,8 @@ fn restore(files: js_sys::Array) {
1054}1054}
10551055
1056/// Writes the files changed since the last time out to the browser's storage.1056/// Writes the files changed since the last time out to the browser's storage.
1057fn store() {1057#[wasm_bindgen]
1058pub fn flush_storage() {
1058 use notebook::fs::Change;1059 use notebook::fs::Change;
1059 STORE_DUE.set(None);1060 STORE_DUE.set(None);
1060 let changes = js_sys::Array::new();1061 let changes = js_sys::Array::new();
...@@ -1489,7 +1490,7 @@ pub fn flush() {...@@ -1489,7 +1490,7 @@ pub fn flush() {
1489 report(error);1490 report(error);
1490 }1491 }
1491 });1492 });
1492 store();1493 flush_storage();
1493}1494}
14941495
1495#[wasm_bindgen]1496#[wasm_bindgen]
...@@ -1585,7 +1586,7 @@ fn turn(state: &mut State) -> Result<(), Box<dyn Error>> {...@@ -1585,7 +1586,7 @@ fn turn(state: &mut State) -> Result<(), Box<dyn Error>> {
1585 STORE_DUE.set(Some(now + STORE_AFTER));1586 STORE_DUE.set(Some(now + STORE_AFTER));
1586 }1587 }
1587 if STORE_DUE.get().is_some_and(|due| due <= now) {1588 if STORE_DUE.get().is_some_and(|due| due <= now) {
1588 store();1589 flush_storage();
1589 }1590 }
1590 let updating = follow_updates(state, now);1591 let updating = follow_updates(state, now);
1591 let next = [1592 let next = [
...@@ -1703,7 +1704,7 @@ pub fn update_now(state: &mut State) {...@@ -1703,7 +1704,7 @@ pub fn update_now(state: &mut State) {
1703 if let Err(error) = state.persist() {1704 if let Err(error) = state.persist() {
1704 report(error);1705 report(error);
1705 }1706 }
1706 store();1707 flush_storage();
1707 let Some(build) = UPDATE.with_borrow(Clone::clone) else {1708 let Some(build) = UPDATE.with_borrow(Clone::clone) else {
1708 return RELOADING.set(false);1709 return RELOADING.set(false);
1709 };1710 };
crates/snowbound/web/glue.js+1-1
...@@ -639,7 +639,7 @@ export function adoptCanvas(fresh) {...@@ -639,7 +639,7 @@ export function adoptCanvas(fresh) {
639/** Wires the page's canvas, text area and file input to `module`'s exports. */639/** Wires the page's canvas, text area and file input to `module`'s exports. */
640export function attach(module) {640export function attach(module) {
641 wasm = module;641 wasm = module;
642 globalThis.snowboundFlushStorage = () => { wasm.flush(); return settled(); };642 globalThis.snowboundFlushStorage = () => { wasm.flush_storage(); return settled(); };
643 canvas = document.getElementById("page");643 canvas = document.getElementById("page");
644 input = document.getElementById("input");644 input = document.getElementById("input");
645 picker = document.getElementById("files");645 picker = document.getElementById("files");
tools/web/test_storage_barrier.mjs created+59
...@@ -0,0 +1,59 @@
1import assert from 'node:assert/strict';
2import { readFile } from 'node:fs/promises';
3import test from 'node:test';
4import vm from 'node:vm';
5
6test('the durability bridge drains files without reentering the editor and waits for ordered worker receipts', async () => {
7 const source = await readFile(new URL('../../crates/snowbound/web/glue.js', import.meta.url), 'utf8');
8 const messages = [];
9 let worker;
10 const element = { style: {}, addEventListener() {}, focus() {}, remove() {} };
11 const context = vm.createContext({
12 Worker: class {
13 constructor() { worker = this; }
14 postMessage(data) {
15 messages.push(data);
16 if (data.kind === 'load') queueMicrotask(() => this.onmessage({ data: [] }));
17 }
18 },
19 Blob,
20 URL,
21 document: { getElementById: () => element, addEventListener() {} },
22 localStorage: { getItem: () => null },
23 ResizeObserver: class { observe() {} },
24 matchMedia: () => ({ addEventListener() {} }),
25 addEventListener() {},
26 console,
27 });
28 vm.runInContext(source.replace(/\bexport /g, '').replaceAll('import.meta.url', JSON.stringify(import.meta.url)), context);
29 await vm.runInContext('loadFiles()', context);
30 const flushed = [];
31 context.module = {
32 flush() { throw Error('Durability reentered State.persist'); },
33 flush_storage() {
34 const marker = flushed.length + 1;
35 flushed.push(marker);
36 context.nextChange = [['replica-wal', marker, [[0, new Uint8Array([marker])]]]];
37 vm.runInContext('storeFiles(nextChange)', context);
38 },
39 };
40 vm.runInContext('attach(module)', context);
41 const replies = [];
42 const first = vm.runInContext('snowboundFlushStorage()', context).then(() => replies.push(1));
43 const second = vm.runInContext('snowboundFlushStorage()', context).then(() => replies.push(2));
44 assert.deepEqual(messages.map(message => message.kind), ['load', 'store', 'settle', 'store', 'settle']);
45 await Promise.resolve();
46 assert.deepEqual(replies, []);
47 worker.onmessage({ data: { error: null } });
48 await first;
49 assert.deepEqual(replies, [1]);
50 worker.onmessage({ data: { error: null } });
51 await second;
52 assert.deepEqual(replies, [1, 2]);
53 const rejected = vm.runInContext('snowboundFlushStorage()', context);
54 worker.onmessage({ data: { error: 'Storage unavailable' } });
55 await assert.rejects(rejected, /Storage unavailable/);
56 const retried = vm.runInContext('snowboundFlushStorage()', context);
57 worker.onmessage({ data: { error: null } });
58 await retried;
59});