1//! The section thread: the only owner of the parsed section, which is the cached base
2//! image with each sealed batch replayed and the open batch's edits applied. Edits apply
3//! here and are written in one SQLite transaction per burst; the sync thread asks it to seal
4//! and, when the remote changed, to rebase the queue. In the browser, which has one thread,
5//! the section is served as each request is sent, and a rebuild runs as it is asked for.
6
7use crate::{Moving, Result, base, lock, merge, queue, worker::Signal};
8use onestore::{
9 Arena, ExGuid, Section, Transaction,
10 op::{Edit, OpError},
11 page::Page,
12 protected::Key,
13};
14use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
15use std::{
16 collections::{BTreeMap, BTreeSet, VecDeque},
17 io,
18 path::PathBuf,
19 sync::{
20 Arc, Mutex, Weak,
21 atomic::{AtomicUsize, Ordering},
22 mpsc,
23 },
24};
25
26pub(crate) type Reply<T> = Box<dyn FnOnce(Result<T>) + Send>;
27pub(crate) type Notice = Box<dyn Fn(&str) + Send>;
28
29pub(crate) enum Request {
30 Apply(Accepted),
31 Page {
32 space: ExGuid,
33 reply: Reply<Page>,
34 },
35 Pages {
36 reply: Reply<Vec<(ExGuid, String, u32)>>,
37 },
38 Conflicts {
39 reply: Reply<Vec<(ExGuid, Vec<onestore::ConflictPage>)>>,
40 },
41 Versions {
42 reply: Reply<Vec<(ExGuid, Vec<onestore::PageVersion>)>>,
43 },
44 Version {
45 space: ExGuid,
46 version: ExGuid,
47 reply: Reply<Page>,
48 },
49 Moving {
50 space: ExGuid,
51 reply: Reply<Moving>,
52 },
53 /// Answers once the edits before it are written, optionally awaiting host durability.
54 Flush {
55 reply: Reply<()>,
56 durable: bool,
57 },
58 /// Seals the open batch, when no sealed batch waits for publication, recording the
59 /// attempt to publish it.
60 Seal {
61 reply: Reply<Option<Sealed>>,
62 },
63 /// Replays the queue on `image` (the stored remote image, or the base, when `None`);
64 /// answers with the pages the remote changed.
65 Rebase {
66 image: Option<Vec<u8>>,
67 reply: Reply<Vec<ExGuid>>,
68 },
69 /// Rereads the queue after the sync thread replaced it.
70 Reopen {
71 reply: Reply<()>,
72 },
73 /// From the thread that rebuilt the section: hand it the requests.
74 #[cfg_attr(target_arch = "wasm32", allow(dead_code))]
75 Handover(mpsc::SyncSender<Takeover>),
76 /// From the thread that rebuilt nothing: carry on with the section as it is.
77 #[cfg_attr(target_arch = "wasm32", allow(dead_code))]
78 Resume,
79}
80
81/// The request channel and the requests held while the section was rebuilt.
82type Takeover = (mpsc::Receiver<Request>, VecDeque<Request>);
83
84/// A sealed batch: the transaction publishing it, or none when its edits changed nothing.
85pub(crate) struct Sealed {
86 pub batch: i64,
87 pub transaction: Option<Transaction>,
88}
89
90impl Request {
91 fn fail(self, kind: io::ErrorKind, message: &str) {
92 let error = || io::Error::new(kind, message.to_owned()).into();
93 match self {
94 Self::Apply(Accepted { reply, .. }) => reply(Err(error())),
95 Self::Page { reply, .. } => reply(Err(error())),
96 Self::Pages { reply } => reply(Err(error())),
97 Self::Conflicts { reply } => reply(Err(error())),
98 Self::Versions { reply } => reply(Err(error())),
99 Self::Version { reply, .. } => reply(Err(error())),
100 Self::Moving { reply, .. } => reply(Err(error())),
101 Self::Flush { reply, .. } => reply(Err(error())),
102 Self::Seal { reply } => reply(Err(error())),
103 Self::Rebase { reply, .. } => reply(Err(error())),
104 Self::Reopen { reply } => reply(Err(error())),
105 Self::Handover(_) | Self::Resume => {}
106 }
107 }
108}
109
110/// The section thread as the replica sees it. Rereading the section after the queue was
111/// replaced (a rebase, a released attempt) happens on a new thread while the current one
112/// keeps answering page reads from the section as it was; the new thread then takes the
113/// requests over, so reads never wait for a rebuild.
114pub(crate) struct Thread {
115 pub(crate) connection: Mutex<Connection>,
116 /// The sync worker, which a durable burst of edits wakes.
117 pub(crate) worker: Mutex<Weak<Signal>>,
118 /// Taken when the replica drops, which ends every section thread.
119 sender: Mutex<Option<mpsc::Sender<Request>>>,
120 #[cfg(target_arch = "wasm32")]
121 durable: Mutex<Option<mpsc::Sender<Receipt>>>,
122 #[cfg(target_arch = "wasm32")]
123 durable_wake: crate::task::Sender<()>,
124 #[cfg(not(target_arch = "wasm32"))]
125 threads: Mutex<Vec<std::thread::JoinHandle<()>>>,
126 /// A password-protected section's key, under which its queue is sealed too.
127 pub(crate) key: Option<Key>,
128 /// The edits the cache last refused for now, as `crate::unwritten` counts them.
129 waiting: AtomicUsize,
130 /// Last, so the cache closes before background synchronization takes it over.
131 pub(crate) released: crate::Released,
132}
133
134impl Drop for Thread {
135 fn drop(&mut self) {
136 self.wait(0);
137 }
138}
139
140impl Thread {
141 fn wait(&self, edits: usize) {
142 let was = self.waiting.swap(edits, Ordering::Relaxed);
143 crate::UNWRITTEN.fetch_add(edits, Ordering::Relaxed);
144 crate::UNWRITTEN.fetch_sub(was, Ordering::Relaxed);
145 }
146
147 pub(crate) fn send(&self, request: Request) -> Result<()> {
148 self.sender
149 .lock()
150 .ok()
151 .and_then(|sender| sender.as_ref()?.send(request).ok())
152 .ok_or_else(|| io::Error::other("The section thread stopped"))?;
153 #[cfg(target_arch = "wasm32")]
154 serving::serve(self);
155 Ok(())
156 }
157
158 /// Ends request serving; pending browser receipts retain the cache until they settle.
159 pub(crate) fn stop(&self) {
160 if let Ok(mut sender) = self.sender.lock() {
161 sender.take();
162 }
163 #[cfg(target_arch = "wasm32")]
164 {
165 serving::serve(self);
166 if let Ok(mut durable) = self.durable.lock() {
167 durable.take();
168 }
169 let _ = self.durable_wake.try_send(());
170 }
171 #[cfg(not(target_arch = "wasm32"))]
172 while let Some(thread) = self
173 .threads
174 .lock()
175 .ok()
176 .and_then(|mut threads| threads.pop())
177 {
178 let _ = thread.join();
179 }
180 }
181
182 #[cfg(not(target_arch = "wasm32"))]
183 fn start<F: Future<Output = ()>>(
184 self: &Arc<Self>,
185 run: impl FnOnce(Arc<Self>) -> F + Send + 'static,
186 ) -> Result<()> {
187 let shared = Arc::clone(self);
188 let thread = std::thread::Builder::new()
189 .name("onestore-section".into())
190 .spawn(move || crate::task::complete(run(shared)))?;
191 self.threads
192 .lock()
193 .map_err(|_| io::Error::other("The section thread panicked"))?
194 .push(thread);
195 Ok(())
196 }
197
198 #[cfg(target_arch = "wasm32")]
199 fn start<F: Future<Output = ()> + 'static>(
200 self: &Arc<Self>,
201 run: impl FnOnce(Arc<Self>) -> F,
202 ) -> Result<()> {
203 serving::start(self, Box::pin(run(Arc::clone(self))));
204 Ok(())
205 }
206
207 #[cfg(target_arch = "wasm32")]
208 fn receipt(self: &Arc<Self>, replies: Vec<Reply<()>>, edited: bool) {
209 let receipt = Receipt {
210 shared: Arc::clone(self),
211 replies,
212 edited,
213 };
214 let sent = match self.durable.lock() {
215 Ok(sender) => match sender.as_ref() {
216 Some(sender) => sender.send(receipt).map_err(|error| error.0),
217 None => Err(receipt),
218 },
219 Err(_) => Err(receipt),
220 };
221 if let Err(receipt) = sent {
222 for reply in receipt.replies {
223 reply(Err(io::Error::other("The section thread stopped").into()));
224 }
225 }
226 let _ = self.durable_wake.try_send(());
227 }
228}
229
230#[cfg(target_arch = "wasm32")]
231struct Receipt {
232 shared: Arc<Thread>,
233 replies: Vec<Reply<()>>,
234 edited: bool,
235}
236
237#[cfg(target_arch = "wasm32")]
238async fn receipts(requests: mpsc::Receiver<Receipt>, wake: crate::task::Receiver<()>) {
239 loop {
240 let Receipt {
241 shared,
242 replies,
243 edited,
244 } = match requests.try_recv() {
245 Ok(receipt) => receipt,
246 Err(mpsc::TryRecvError::Empty) => {
247 crate::task::wait(&wake, None).await;
248 continue;
249 }
250 Err(mpsc::TryRecvError::Disconnected) => return,
251 };
252 let result = loop {
253 match crate::fs::durable().await {
254 Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
255 crate::task::wait(&wake, Some(std::time::Duration::from_secs(1))).await;
256 }
257 result => break result,
258 }
259 };
260 match result {
261 Ok(()) => {
262 for reply in replies {
263 reply(Ok(()));
264 }
265 if edited {
266 crate::edited(&shared.worker);
267 }
268 }
269 Err(error) => {
270 let message = error.to_string();
271 for reply in replies {
272 reply(Err(io::Error::new(error.kind(), message.clone()).into()));
273 }
274 }
275 }
276 }
277}
278
279/// The browser's section threads: each replica's `run`, polled whenever a request is sent
280/// until it waits for the next.
281#[cfg(target_arch = "wasm32")]
282mod serving {
283 use super::Thread;
284 use std::{cell::RefCell, collections::BTreeMap, future::Future, pin::Pin, task};
285
286 type Serving = Pin<Box<dyn Future<Output = ()>>>;
287
288 thread_local! {
289 /// By the address of each replica's `Thread`, which its `run` keeps alive.
290 static SERVING: RefCell<BTreeMap<usize, Serving>> = const { RefCell::new(BTreeMap::new()) };
291 }
292
293 pub(super) fn start(thread: &Thread, run: Serving) {
294 SERVING.with_borrow_mut(|serving| serving.insert(std::ptr::from_ref(thread).addr(), run));
295 serve(thread);
296 }
297
298 /// Serves what `thread` was sent. A request sent while it is being served, as from a
299 /// reply, waits for the serving already under way, which takes it next.
300 pub(super) fn serve(thread: &Thread) {
301 let id = std::ptr::from_ref(thread).addr();
302 let Some(mut run) = SERVING.with_borrow_mut(|serving| serving.remove(&id)) else {
303 return;
304 };
305 let mut context = task::Context::from_waker(task::Waker::noop());
306 if run.as_mut().poll(&mut context).is_pending() {
307 SERVING.with_borrow_mut(|serving| serving.insert(id, run));
308 }
309 }
310}
311
312/// The next request; a second passes without one while `retry`.
313#[cfg(not(target_arch = "wasm32"))]
314async fn receive(
315 requests: &mpsc::Receiver<Request>,
316 retry: bool,
317) -> std::result::Result<Request, mpsc::RecvTimeoutError> {
318 if retry {
319 requests.recv_timeout(std::time::Duration::from_secs(1))
320 } else {
321 requests
322 .recv()
323 .map_err(|_| mpsc::RecvTimeoutError::Disconnected)
324 }
325}
326
327/// The next request, waiting for `serving::serve` where none is queued; requests alone
328/// retry unwritten edits.
329#[cfg(target_arch = "wasm32")]
330async fn receive(
331 requests: &mpsc::Receiver<Request>,
332 _retry: bool,
333) -> std::result::Result<Request, mpsc::RecvTimeoutError> {
334 std::future::poll_fn(|_| match requests.try_recv() {
335 Ok(request) => std::task::Poll::Ready(Ok(request)),
336 Err(mpsc::TryRecvError::Disconnected) => {
337 std::task::Poll::Ready(Err(mpsc::RecvTimeoutError::Disconnected))
338 }
339 Err(mpsc::TryRecvError::Empty) => std::task::Poll::Pending,
340 })
341 .await
342}
343
344/// Starts the section thread once the queue opens.
345pub(crate) fn spawn(connection: Connection, key: Option<Key>) -> Result<(Arc<Thread>, ExGuid)> {
346 let (sender, requests) = mpsc::channel();
347 #[cfg(target_arch = "wasm32")]
348 let (durable, written) = mpsc::channel();
349 #[cfg(target_arch = "wasm32")]
350 let (durable_wake, woken) = crate::task::channel();
351 let shared = Arc::new(Thread {
352 connection: Mutex::new(connection),
353 worker: Mutex::new(Weak::new()),
354 sender: Mutex::new(Some(sender)),
355 #[cfg(target_arch = "wasm32")]
356 durable: Mutex::new(Some(durable)),
357 #[cfg(target_arch = "wasm32")]
358 durable_wake,
359 #[cfg(not(target_arch = "wasm32"))]
360 threads: Mutex::new(Vec::new()),
361 key,
362 waiting: AtomicUsize::new(0),
363 released: crate::Released::default(),
364 });
365 #[cfg(target_arch = "wasm32")]
366 crate::task::spawn("onestore-durable", move || receipts(written, woken))?;
367 let (ready, opened) = mpsc::sync_channel(1);
368 shared.start(move |shared| run(shared, requests, VecDeque::new(), Some(ready)))?;
369 match opened.recv() {
370 Ok(Ok(root)) => Ok((shared, root)),
371 Ok(Err(error)) => {
372 shared.stop();
373 Err(error)
374 }
375 Err(_) => {
376 shared.stop();
377 Err(io::Error::other("The section thread panicked").into())
378 }
379 }
380}
381
382enum Next {
383 Stop,
384 Reopen,
385 Handover(mpsc::SyncSender<Takeover>, VecDeque<Request>),
386}
387
388async fn run(
389 shared: Arc<Thread>,
390 requests: mpsc::Receiver<Request>,
391 mut backlog: VecDeque<Request>,
392 mut ready: Option<mpsc::SyncSender<Result<ExGuid>>>,
393) {
394 loop {
395 let arena = Arena::default();
396 let working = match Working::open(&arena, &shared.connection, shared.key.as_ref()) {
397 Ok(working) => working,
398 Err(error) => {
399 if let Some(ready) = ready.take() {
400 let _ = ready.send(Err(error));
401 } else {
402 fail(&shared, error, backlog, requests);
403 }
404 return;
405 }
406 };
407 if let Some(ready) = ready.take() {
408 let _ = ready.send(Ok(working.section.root()));
409 }
410 match working.serve(&shared, &requests, &mut backlog).await {
411 Next::Stop => return,
412 Next::Reopen => {}
413 Next::Handover(to, held) => {
414 let _ = to.send((requests, held));
415 return;
416 }
417 }
418 }
419}
420
421/// Answers every request with `error` until the replica drops.
422fn fail(
423 shared: &Thread,
424 error: crate::Error,
425 backlog: VecDeque<Request>,
426 requests: mpsc::Receiver<Request>,
427) {
428 // The edits that waited are in the backlog, refused with the rest.
429 shared.wait(0);
430 let message = error.to_string();
431 for request in backlog.into_iter().chain(requests.iter()) {
432 request.fail(io::ErrorKind::Other, &message);
433 }
434}
435
436/// What a rebuilding thread does before it rereads the section.
437#[cfg_attr(target_arch = "wasm32", allow(dead_code))]
438enum Job {
439 Rebase {
440 image: Option<Vec<u8>>,
441 reply: Reply<Vec<ExGuid>>,
442 },
443 Reopen {
444 reply: Reply<()>,
445 },
446}
447
448/// Runs `job`, rereads the section and takes the requests over from the thread that
449/// started it; a failed rebase changed nothing that thread serves.
450#[cfg(not(target_arch = "wasm32"))]
451fn build(shared: Arc<Thread>, job: Job, signal: mpsc::Sender<Request>) {
452 let answer: Box<dyn FnOnce() + Send> = match job {
453 Job::Reopen { reply } => Box::new(move || reply(Ok(()))),
454 Job::Rebase { image, reply } => {
455 match rebase(&shared.connection, image, shared.key.as_ref()) {
456 Ok(changed) => Box::new(move || reply(Ok(changed))),
457 Err(error) => {
458 reply(Err(error));
459 let _ = signal.send(Request::Resume);
460 return;
461 }
462 }
463 }
464 };
465 let arena = Arena::default();
466 let opened = Working::open(&arena, &shared.connection, shared.key.as_ref());
467 let (to, from) = mpsc::sync_channel(1);
468 if signal.send(Request::Handover(to)).is_err() {
469 return;
470 }
471 drop(signal);
472 answer();
473 let Ok((requests, mut backlog)) = from.recv() else {
474 return;
475 };
476 let working = match opened {
477 Ok(working) => working,
478 Err(error) => return fail(&shared, error, backlog, requests),
479 };
480 match crate::task::complete(working.serve(&shared, &requests, &mut backlog)) {
481 Next::Stop => {}
482 Next::Reopen => crate::task::complete(run(shared, requests, backlog, None)),
483 Next::Handover(to, held) => {
484 let _ = to.send((requests, held));
485 }
486 }
487}
488
489/// `build` where there is one thread: runs `job` before the section is reread, which the
490/// serving thread does once true is returned; a failed rebase changed nothing it serves.
491#[cfg(target_arch = "wasm32")]
492fn inline(shared: &Thread, job: Job) -> bool {
493 match job {
494 Job::Reopen { reply } => reply(Ok(())),
495 Job::Rebase { image, reply } => {
496 match rebase(&shared.connection, image, shared.key.as_ref()) {
497 Ok(changed) => reply(Ok(changed)),
498 Err(error) => {
499 reply(Err(error));
500 return false;
501 }
502 }
503 }
504 }
505 true
506}
507
508/// Writes again each second the edits a dropped replica left waiting, until none do or their
509/// cache is gone, keeping the cache open and `Released` from firing meanwhile.
510#[cfg(not(target_arch = "wasm32"))]
511fn orphan(shared: &Arc<Thread>, unwritten: Vec<Accepted>) {
512 let shared = Arc::clone(shared);
513 let path = lock(&shared.connection)
514 .ok()
515 .and_then(|connection| connection.path().map(PathBuf::from));
516 let spawned = std::thread::Builder::new()
517 .name("onestore-unwritten".into())
518 .spawn(move || {
519 let arena = Arena::default();
520 let Ok(mut working) = Working::open(&arena, &shared.connection, shared.key.as_ref())
521 else {
522 return;
523 };
524 working.unwritten = unwritten;
525 // A cache deleted, as with its notebook, takes no edits.
526 while !working.unwritten.is_empty() && path.as_ref().is_some_and(|path| path.exists()) {
527 std::thread::sleep(std::time::Duration::from_secs(1));
528 let _ = working.flush(&shared);
529 }
530 });
531 if let Err(error) = spawned {
532 eprintln!("Edits waiting for the cache are lost: {error}");
533 }
534}
535
536/// An edit to apply, with who waits for it.
537pub(crate) struct Accepted {
538 pub author: String,
539 pub edit: Edit,
540 pub reply: Reply<u64>,
541 /// Set where nobody blocks on the reply: a failed write keeps the edit and reports here.
542 pub waiting: Option<Notice>,
543 /// A page as a move copied it, which the edit applies only while it still holds.
544 pub moving: Option<Box<Moving>>,
545}
546
547struct Working<'a> {
548 section: Section<'a>,
549 /// The batch collecting edits; sealed batches before it are replayed.
550 open: Option<i64>,
551 /// Spaces the open batch's edits change.
552 touched: BTreeSet<ExGuid>,
553 /// Applied to the section and not yet in the queue, so the section is the queue replayed
554 /// with these after it; written in order before anything seals, rebases or rereads.
555 unwritten: Vec<Accepted>,
556}
557
558impl<'a> Working<'a> {
559 fn open(arena: &'a Arena, connection: &Mutex<Connection>, key: Option<&Key>) -> Result<Self> {
560 let (section, open, touched) = replay(arena, &*lock(connection)?, key)?;
561 Ok(Self {
562 section,
563 open,
564 touched,
565 unwritten: Vec::new(),
566 })
567 }
568
569 async fn serve(
570 mut self,
571 shared: &Arc<Thread>,
572 requests: &mpsc::Receiver<Request>,
573 backlog: &mut VecDeque<Request>,
574 ) -> Next {
575 let connection = &shared.connection;
576 // While another thread rebuilds the section, reads answer from this one and every
577 // other request waits for the rebuilt section.
578 let mut held: Option<VecDeque<Request>> = None;
579 loop {
580 let mut burst = std::mem::take(backlog);
581 // Only a burst that took an edit, or a retry's tick, writes at its end: reads never.
582 let mut write = false;
583 if burst.is_empty() {
584 match receive(requests, !self.unwritten.is_empty()).await {
585 Ok(request) => burst.push_back(request),
586 Err(mpsc::RecvTimeoutError::Timeout) => write = true,
587 Err(mpsc::RecvTimeoutError::Disconnected) => {
588 let _ = self.flush(shared);
589 #[cfg(not(target_arch = "wasm32"))]
590 if !self.unwritten.is_empty() {
591 orphan(shared, std::mem::take(&mut self.unwritten));
592 }
593 return Next::Stop;
594 }
595 }
596 }
597 burst.extend(requests.try_iter());
598 // Requests that write the queue beyond the burst's edits run after its reads.
599 let mut later = VecDeque::new();
600 let mut reopen = false;
601 while !reopen && let Some(request) = burst.pop_front().or_else(|| later.pop_front()) {
602 if let Some(waiting) = &mut held {
603 match request {
604 Request::Page { space, reply } => {
605 reply(self.section.page(space).map_err(Into::into))
606 }
607 Request::Pages { reply } => reply(self.section.pages().map_err(Into::into)),
608 Request::Conflicts { reply } => {
609 reply(self.section.conflicts().map_err(Into::into))
610 }
611 Request::Versions { reply } => {
612 reply(self.section.versions().map_err(Into::into))
613 }
614 Request::Version {
615 space,
616 version,
617 reply,
618 } => reply(self.section.version(space, version).map_err(Into::into)),
619 Request::Moving { space, reply } => reply(self.moving(space)),
620 Request::Handover(to) => {
621 let mut waiting = held.take().unwrap_or_default();
622 waiting.extend(burst.drain(..).chain(later.drain(..)));
623 return Next::Handover(to, waiting);
624 }
625 Request::Resume => {
626 let waiting = held.take().unwrap_or_default();
627 burst = waiting.into_iter().chain(burst.drain(..)).collect();
628 }
629 other => waiting.push_back(other),
630 }
631 continue;
632 }
633 if matches!(request, Request::Flush { .. } | Request::Seal { .. })
634 && !burst.is_empty()
635 {
636 later.push_back(request);
637 continue;
638 }
639 if matches!(
640 request,
641 Request::Flush { .. } | Request::Seal { .. } | Request::Rebase { .. }
642 ) {
643 let failed = self.flush(shared).is_err();
644 if failed || !self.unwritten.is_empty() {
645 // The worker retries later; the edits wait here meanwhile.
646 request.fail(io::ErrorKind::WouldBlock, "Edits wait to be written");
647 reopen = failed;
648 continue;
649 }
650 }
651 reopen = match request {
652 Request::Apply(accepted) => {
653 write = true;
654 self.accept(accepted)
655 }
656 Request::Page { space, reply } => {
657 reply(self.section.page(space).map_err(Into::into));
658 false
659 }
660 Request::Pages { reply } => {
661 reply(self.section.pages().map_err(Into::into));
662 false
663 }
664 Request::Conflicts { reply } => {
665 reply(self.section.conflicts().map_err(Into::into));
666 false
667 }
668 Request::Versions { reply } => {
669 reply(self.section.versions().map_err(Into::into));
670 false
671 }
672 Request::Version {
673 space,
674 version,
675 reply,
676 } => {
677 reply(self.section.version(space, version).map_err(Into::into));
678 false
679 }
680 Request::Moving { space, reply } => {
681 reply(self.moving(space));
682 false
683 }
684 Request::Flush { reply, durable } => {
685 #[cfg(target_arch = "wasm32")]
686 if durable {
687 shared.receipt(vec![reply], false);
688 } else {
689 reply(Ok(()));
690 }
691 #[cfg(not(target_arch = "wasm32"))]
692 {
693 let _ = durable;
694 reply(Ok(()));
695 }
696 false
697 }
698 Request::Seal { reply } => {
699 let sealed = self.seal(connection);
700 let failed = sealed.is_err();
701 reply(sealed);
702 failed
703 }
704 Request::Rebase { image, reply } => {
705 #[cfg(target_arch = "wasm32")]
706 let reopen = inline(shared, Job::Rebase { image, reply });
707 #[cfg(not(target_arch = "wasm32"))]
708 let reopen = {
709 held = self
710 .rebuild(shared, Job::Rebase { image, reply })
711 .then(VecDeque::new);
712 false
713 };
714 reopen
715 }
716 Request::Reopen { reply } => {
717 let _ = self.flush(shared);
718 #[cfg(target_arch = "wasm32")]
719 let reopen = inline(shared, Job::Reopen { reply });
720 #[cfg(not(target_arch = "wasm32"))]
721 let reopen = {
722 held = self
723 .rebuild(shared, Job::Reopen { reply })
724 .then(|| self.unwritten.drain(..).map(Request::Apply).collect());
725 false
726 };
727 reopen
728 }
729 Request::Handover(_) | Request::Resume => false,
730 };
731 }
732 if reopen || write && self.flush(shared).is_err() {
733 backlog.extend(
734 self.unwritten
735 .drain(..)
736 .map(Request::Apply)
737 .chain(burst.drain(..))
738 .chain(later.drain(..)),
739 );
740 return Next::Reopen;
741 }
742 }
743 }
744
745 /// Starts a thread that runs `job` and rereads the section; false when the replica is
746 /// stopping, which answers the job's request.
747 #[cfg(not(target_arch = "wasm32"))]
748 fn rebuild(&self, shared: &Arc<Thread>, job: Job) -> bool {
749 let signal = shared
750 .sender
751 .lock()
752 .ok()
753 .and_then(|sender| sender.as_ref().cloned());
754 let Some(signal) = signal else {
755 match job {
756 Job::Rebase { reply, .. } => {
757 reply(Err(io::Error::other("The section thread stopped").into()))
758 }
759 Job::Reopen { reply } => {
760 reply(Err(io::Error::other("The section thread stopped").into()))
761 }
762 }
763 return false;
764 };
765 shared
766 .start(move |shared| async move { build(shared, job, signal) })
767 .is_ok()
768 }
769
770 /// Page `space` as a move copies it, with the conflict pages and versions it leaves.
771 fn moving(&mut self, space: ExGuid) -> Result<Moving> {
772 fn of<T>(space: ExGuid, listed: Vec<(ExGuid, Vec<T>)>) -> Vec<T> {
773 (listed.into_iter())
774 .find_map(|(page, of)| (page == space).then_some(of))
775 .unwrap_or_default()
776 }
777 Ok(Moving {
778 space,
779 page: self.section.page(space)?,
780 conflicts: of(space, self.section.conflicts()?),
781 versions: of(space, self.section.versions()?),
782 })
783 }
784
785 /// Applies an edit, keeping it to be written with the burst. A refused edit is answered
786 /// here; true when it failed part way and the section must be reread.
787 fn accept(&mut self, accepted: Accepted) -> bool {
788 if let Some(moving) = &accepted.moving
789 && !self.moving(moving.space).is_ok_and(|now| now == **moving)
790 {
791 let changed = OpError::StructureChanged("The page changed while it moved.");
792 (accepted.reply)(Err(changed.into()));
793 return false;
794 }
795 match self.section.apply(&accepted.author, &accepted.edit) {
796 Ok(()) => {
797 self.touched
798 .extend(queue::spaces(&accepted.edit, self.section.root()));
799 self.unwritten.push(accepted);
800 false
801 }
802 Err(error) => {
803 let broken = matches!(error, OpError::Failed(_));
804 (accepted.reply)(Err(error.into()));
805 broken
806 }
807 }
808 }
809
810 /// Writes the unwritten edits in one transaction, then answers their senders. A write
811 /// that may succeed later keeps the edits with a notice, telling the first; one that never
812 /// will refuses the edit it failed on. An error means the section holds an edit the queue
813 /// never will, or the cache closed, and must be reread.
814 fn flush(&mut self, shared: &Arc<Thread>) -> Result<()> {
815 if self.unwritten.is_empty() {
816 return Ok(());
817 }
818 let mut connection = lock(&shared.connection)?;
819 let mut ids = Vec::new();
820 let committed = connection
821 .transaction_with_behavior(TransactionBehavior::Immediate)
822 .map_err(crate::Error::from)
823 .and_then(|transaction| {
824 let batch = match self.open {
825 Some(batch) => batch,
826 None => {
827 transaction.execute("INSERT INTO batches DEFAULT VALUES", [])?;
828 transaction.last_insert_rowid()
829 }
830 };
831 for edit in &self.unwritten {
832 ids.push(queue::insert(
833 &transaction,
834 shared.key.as_ref(),
835 None,
836 batch,
837 &edit.author,
838 &edit.edit,
839 )?);
840 }
841 Ok((batch, transaction.commit().err()))
842 });
843 // The error, and the edit it refuses when it would fail again.
844 let written = match committed {
845 Ok((batch, None)) => Ok(batch),
846 Ok((batch, Some(error))) => {
847 if landed(&mut connection, ids.last().copied().unwrap_or_default())? {
848 Ok(batch)
849 } else {
850 Err((error.into(), None))
851 }
852 }
853 Err(error) => {
854 let refused = (!transient(&error)).then_some(ids.len());
855 Err((error, refused))
856 }
857 };
858 drop(connection);
859 let batch = match written {
860 Ok(batch) => batch,
861 Err((error, refused)) => {
862 let message = error.to_string();
863 let before = self.unwritten.len();
864 let mut at = 0;
865 for edit in self.unwritten.extract_if(.., |edit| {
866 at += 1;
867 edit.waiting.is_none() || refused == Some(at - 1)
868 }) {
869 (edit.reply)(Err(io::Error::other(message.clone()).into()));
870 }
871 shared.wait(self.unwritten.len());
872 if let Some(notice) = self
873 .unwritten
874 .first()
875 .and_then(|edit| edit.waiting.as_ref())
876 {
877 notice(&message);
878 }
879 return if self.unwritten.len() == before {
880 Ok(())
881 } else {
882 Err(error)
883 };
884 }
885 };
886 self.open = Some(batch);
887 #[cfg(target_arch = "wasm32")]
888 shared.receipt(
889 self.unwritten
890 .drain(..)
891 .zip(ids)
892 .map(|(edit, id)| {
893 let reply: Reply<()> =
894 Box::new(move |result| (edit.reply)(result.map(|()| id)));
895 reply
896 })
897 .collect(),
898 true,
899 );
900 #[cfg(not(target_arch = "wasm32"))]
901 {
902 for (edit, id) in self.unwritten.drain(..).zip(ids) {
903 (edit.reply)(Ok(id));
904 }
905 crate::edited(&shared.worker);
906 }
907 shared.wait(0);
908 Ok(())
909 }
910
911 /// Seals the open batch unless a sealed batch still waits for publication.
912 fn seal(&mut self, connection: &Mutex<Connection>) -> Result<Option<Sealed>> {
913 let Some(batch) = self.open else {
914 return Ok(None);
915 };
916 {
917 let connection = lock(connection)?;
918 let waiting: bool = connection.query_row(
919 "SELECT EXISTS(SELECT 1 FROM batches WHERE sealed IS NOT NULL)",
920 [],
921 |row| row.get(0),
922 )?;
923 if waiting {
924 return Ok(None);
925 }
926 }
927 let transaction = self.section.seal()?;
928 // A batch whose edits change nothing is recorded against the section's root.
929 let root = self.section.root();
930 let revisions: BTreeMap<ExGuid, ExGuid> = self
931 .section
932 .newest()
933 .filter(|(space, _)| {
934 self.touched.contains(space) || (self.touched.is_empty() && *space == root)
935 })
936 .collect();
937 // The sync thread publishes what it seals at once: the attempt is recorded with it.
938 lock(connection)?.execute(
939 "UPDATE batches SET sealed=?1, revisions=?2, attempted=?3 WHERE id=?4",
940 params![
941 serde_json::to_string(&transaction).map_err(io::Error::other)?,
942 serde_json::to_string(&revisions).map_err(io::Error::other)?,
943 transaction.is_some(),
944 batch
945 ],
946 )?;
947 self.open = None;
948 self.touched.clear();
949 Ok(Some(Sealed { batch, transaction }))
950 }
951}
952
953/// Replays every queued edit on `image`, which becomes the base; each page whose local
954/// version the remote could not take gains a conflict page holding it, queued as a new edit.
955/// Returns the pages the remote changed.
956fn rebase(
957 connection: &Mutex<Connection>,
958 image: Option<Vec<u8>>,
959 key: Option<&Key>,
960) -> Result<Vec<ExGuid>> {
961 let (base_image, image, edits) = {
962 let connection = lock(connection)?;
963 if connection.query_row(
964 "SELECT EXISTS(SELECT 1 FROM batches WHERE attempted=1)",
965 [],
966 |row| row.get::<_, bool>(0),
967 )? {
968 return Err(io::Error::new(
969 io::ErrorKind::InvalidInput,
970 "An uncertain attempt is never rebased",
971 )
972 .into());
973 }
974 let base_image = base::base(&connection)?;
975 let image = match image {
976 Some(image) => image,
977 None => {
978 base::read(&connection, base::Image::Remote)?.unwrap_or_else(|| base_image.clone())
979 }
980 };
981 (base_image, image, queue::load(&connection, key, None)?)
982 };
983 let old_arena = Arena::default();
984 let mut old = open(&old_arena, base_image, key)?;
985 let before: BTreeMap<ExGuid, ExGuid> = old.revisions().collect();
986 let remote_arena = Arena::default();
987 let remote = open(&remote_arena, image.clone(), key)?;
988 if old.root() != remote.root() {
989 return Err(io::Error::new(
990 io::ErrorKind::InvalidInput,
991 "Remote snapshot belongs to another document",
992 )
993 .into());
994 }
995 let after: BTreeMap<ExGuid, ExGuid> = remote.revisions().collect();
996 // The pages as the queue leaves them, read once a page conflicts: O(section).
997 let local_arena = Arena::default();
998 let mut local = None;
999 // Pages the remote already holds as the local edits leave them: their ops are done.
1000 let mut converged = BTreeSet::new();
1001 let (rewritten, added) = loop {
1002 let arena = Arena::default();
1003 let mut new = open(&arena, image.clone(), key)?;
1004 let merged = merge::rebase(&mut old, &mut new, &edits, &converged)?;
1005 if !merged.conflicts.is_empty() && local.is_none() {
1006 local = Some(replay(&local_arena, &*lock(connection)?, key)?.0);
1007 }
1008 let pages: BTreeMap<ExGuid, Page> = merged
1009 .conflicts
1010 .keys()
1011 .filter_map(|space| Some((*space, local.as_ref()?.page(*space).ok()?)))
1012 .collect();
1013 let settled: Vec<ExGuid> = pages
1014 .iter()
1015 .filter(|(space, page)| remote.page(**space).is_ok_and(|remote| remote == **page))
1016 .map(|(space, _)| *space)
1017 .collect();
1018 if !settled.is_empty() {
1019 converged.extend(settled);
1020 continue;
1021 }
1022 let mut added = Vec::new();
1023 for (space, (author, objects)) in &merged.conflicts {
1024 // A page the queue went on to delete keeps no version of its own.
1025 let (Some(page), Some(local)) = (pages.get(space), local.as_mut()) else {
1026 continue;
1027 };
1028 let edit = merge::conflict_page(
1029 &mut new,
1030 local,
1031 &merged.moved,
1032 *space,
1033 page,
1034 author,
1035 objects,
1036 crate::now(),
1037 None,
1038 )?;
1039 added.push((author.clone(), edit));
1040 }
1041 break (merged.rewritten, added);
1042 };
1043 let mut connection = lock(connection)?;
1044 let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
1045 base::write(&transaction, base::Image::Base, &image)?;
1046 base::clear(&transaction, base::Image::Remote)?;
1047 transaction.execute("INSERT INTO batches DEFAULT VALUES", [])?;
1048 let batch = transaction.last_insert_rowid();
1049 transaction.execute("UPDATE edits SET batch=?1", [batch])?;
1050 transaction.execute("DELETE FROM batches WHERE id<>?1", [batch])?;
1051 for (id, edit) in rewritten {
1052 queue::rewrite(&transaction, key, id, &edit)?;
1053 }
1054 for (author, edit) in &added {
1055 queue::insert(&transaction, key, None, batch, author, edit)?;
1056 }
1057 if transaction.query_row("SELECT count(*) FROM edits", [], |row| row.get::<_, i64>(0))? == 0 {
1058 transaction.execute("DELETE FROM batches", [])?;
1059 }
1060 queue::collect(&transaction)?;
1061 transaction.commit()?;
1062 let mut changed: BTreeSet<ExGuid> = before
1063 .iter()
1064 .filter(|(space, rid)| after.get(space) != Some(rid))
1065 .map(|(space, _)| *space)
1066 .chain(
1067 after
1068 .keys()
1069 .filter(|space| !before.contains_key(space))
1070 .copied(),
1071 )
1072 .collect();
1073 changed.extend(converged);
1074 for (space, rid) in &before {
1075 if after.get(space) == Some(rid)
1076 && let (Ok(before), Ok(after)) = (old.page(*space), remote.page(*space))
1077 && before != after
1078 {
1079 changed.insert(*space);
1080 }
1081 }
1082 Ok(changed.into_iter().collect())
1083}
1084
1085/// Batches in order with their sealed transactions: `None` while open, `Some(None)` when
1086/// sealing stored nothing.
1087fn batches(connection: &Connection) -> Result<Vec<(i64, Option<Option<Transaction>>)>> {
1088 let mut query = connection.prepare("SELECT id, sealed FROM batches ORDER BY id")?;
1089 let mut rows = query.query([])?;
1090 let mut batches = Vec::new();
1091 while let Some(row) = rows.next()? {
1092 let sealed: Option<String> = row.get(1)?;
1093 batches.push((
1094 row.get(0)?,
1095 sealed
1096 .map(|sealed| serde_json::from_str(&sealed))
1097 .transpose()
1098 .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?,
1099 ));
1100 }
1101 Ok(batches)
1102}
1103
1104/// The cached base with each sealed batch replayed and the open batch's edits applied,
1105/// with the open batch and the spaces its edits change.
1106fn replay<'a>(
1107 arena: &'a Arena,
1108 connection: &Connection,
1109 key: Option<&Key>,
1110) -> Result<(Section<'a>, Option<i64>, BTreeSet<ExGuid>)> {
1111 let image = base::base(connection)?;
1112 let mut section = open(arena, image, key)?;
1113 let mut open = None;
1114 let mut touched = BTreeSet::new();
1115 for (batch, sealed) in batches(connection)? {
1116 if open.is_some() {
1117 return Err(io::Error::new(
1118 io::ErrorKind::InvalidData,
1119 "A cached batch follows the open batch",
1120 )
1121 .into());
1122 }
1123 match sealed {
1124 Some(Some(transaction)) => section.replay(&transaction)?,
1125 Some(None) => {}
1126 None => {
1127 for queued in queue::load(connection, key, Some(batch))? {
1128 section
1129 .apply(&queued.author, &queued.edit)
1130 .map_err(|error| {
1131 io::Error::new(
1132 io::ErrorKind::InvalidData,
1133 format!("Queued edit {} no longer applies: {error}", queued.id),
1134 )
1135 })?;
1136 touched.extend(queue::spaces(&queued.edit, section.root()));
1137 }
1138 open = Some(batch);
1139 }
1140 }
1141 }
1142 Ok((section, open, touched))
1143}
1144
1145/// The image the queue leaves, its unsealed edits sealed as one more revision; that
1146/// revision's identities are fresh on every call. O(section), for tests and recovery.
1147pub(crate) fn image(connection: &Connection, key: Option<&Key>) -> Result<Vec<u8>> {
1148 let arena = Arena::default();
1149 let (mut section, ..) = replay(&arena, connection, key)?;
1150 section.seal()?;
1151 Ok(section.image())
1152}
1153
1154/// A section image, a password-protected one under `key`.
1155pub(crate) fn open<'a>(arena: &'a Arena, image: Vec<u8>, key: Option<&Key>) -> Result<Section<'a>> {
1156 Ok(match key {
1157 Some(key) => Section::unlock(arena, image, key)?,
1158 None => Section::open(arena, image)?,
1159 })
1160}
1161
1162/// The sealed batch waiting for publication, if any.
1163pub(crate) fn sealed(connection: &Connection) -> Result<Option<Sealed>> {
1164 let row: Option<(i64, String)> = connection
1165 .query_row(
1166 "SELECT id, sealed FROM batches WHERE sealed IS NOT NULL ORDER BY id LIMIT 1",
1167 [],
1168 |row| Ok((row.get(0)?, row.get(1)?)),
1169 )
1170 .optional()?;
1171 row.map(|(batch, sealed)| {
1172 Ok(Sealed {
1173 batch,
1174 transaction: serde_json::from_str(&sealed)
1175 .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?,
1176 })
1177 })
1178 .transpose()
1179}
1180
1181/// Whether a write that failed may succeed when tried again, once the disk has room or a lock
1182/// is let go; any other error would fail the same way forever.
1183fn transient(error: &crate::Error) -> bool {
1184 use rusqlite::ErrorCode::*;
1185 matches!(error, crate::Error::Database(rusqlite::Error::SqliteFailure(error, _))
1186 if matches!(error.code, SystemIoFailure | DiskFull | DatabaseBusy | DatabaseLocked | CannotOpen))
1187}
1188
1189/// Whether the commit whose last edit is `last` reached the cache after it failed, as a cold
1190/// open reads it, a commit landing whole: the warm connection may not know of frames its WAL
1191/// holds, and must never write again. When the cache cannot be reopened, an inert connection
1192/// stays, failing every statement.
1193fn landed(connection: &mut Connection, last: u64) -> Result<bool> {
1194 let path = connection.path().map(PathBuf::from);
1195 let inert = Connection::open_in_memory()?;
1196 inert.execute_batch("PRAGMA query_only=1")?;
1197 std::mem::replace(connection, inert)
1198 .close()
1199 .map_err(|(_, error)| error)?;
1200 // Reopening through the browser's file system is unverified: it stays closed.
1201 let path = path
1202 .filter(|_| cfg!(not(target_arch = "wasm32")))
1203 .ok_or_else(|| io::Error::other("The cache could not confirm a write"))?;
1204 let cold = crate::cache_connection(&path)?;
1205 crate::write_ahead(&cold)?;
1206 let found: bool = cold.query_row(
1207 "SELECT EXISTS(SELECT 1 FROM edits WHERE id=?1)",
1208 [crate::signed(last)?],
1209 |row| row.get(0),
1210 )?;
1211 *connection = cold;
1212 Ok(found)
1213}