1#[cfg(target_arch = "wasm32")]
2#[path = "../tests/support/server.rs"]
3mod server;
4
5#[cfg(target_arch = "wasm32")]
6mod web {
7 use super::server::{Server, text};
8 use notebook::{Remote, Replica};
9 use onestore::{
10 CommitError, Stamp, Transaction,
11 op::{Edit, Op, PageOp},
12 };
13 use std::{
14 future::Future,
15 io,
16 sync::{
17 Arc, Mutex,
18 atomic::{AtomicUsize, Ordering},
19 },
20 task::{Context, Poll, Waker},
21 time::Duration,
22 };
23 use wasm_bindgen::prelude::*;
24
25 #[wasm_bindgen]
26 extern "C" {
27 #[wasm_bindgen(js_namespace = console, js_name = error)]
28 fn report_failure(message: &str);
29 #[wasm_bindgen(js_namespace = globalThis, js_name = snowboundReceiptFailure)]
30 fn fail_storage(mode: u32);
31 #[wasm_bindgen(js_namespace = globalThis, js_name = snowboundReceiptTurn)]
32 fn turn() -> js_sys::Promise;
33 #[wasm_bindgen(js_namespace = globalThis, js_name = snowboundReceiptPhase)]
34 fn phase(message: &str);
35 }
36
37 async fn next_turn() {
38 wasm_bindgen_futures::JsFuture::from(turn()).await.unwrap();
39 }
40
41 #[derive(Clone)]
42 struct Shared(Arc<Mutex<Server>>);
43
44 impl Remote for Shared {
45 fn read(&mut self) -> io::Result<Vec<u8>> {
46 self.0.lock().unwrap().read()
47 }
48 fn stamp(&mut self) -> io::Result<Stamp> {
49 self.0.lock().unwrap().stamp()
50 }
51 fn publish(&mut self, transaction: &Transaction) -> Result<(), CommitError> {
52 self.0.lock().unwrap().publish(transaction)
53 }
54 fn confirm(&mut self, base: &Stamp) -> Result<(), CommitError> {
55 self.0.lock().unwrap().confirm(base)
56 }
57 }
58
59 #[wasm_bindgen]
60 pub async fn receipt_probe() {
61 std::panic::set_hook(Box::new(|failure| report_failure(&failure.to_string())));
62 phase("opening idle worker");
63 notebook::fs::create_dir_all("/receipts").unwrap();
64 let source = onestore::create_section("Receipt.one", "original", "Fixture").unwrap();
65 let (space, text_id, _) = text(&source);
66 let unix = web_time::SystemTime::now()
67 .duration_since(web_time::UNIX_EPOCH)
68 .unwrap();
69 let at =
70 (unix.as_secs() + 11_644_473_600) * 10_000_000 + u64::from(unix.subsec_nanos() / 100);
71 let edit = || Edit {
72 at,
73 ops: vec![Op::Page {
74 space,
75 op: PageOp::Text {
76 text: text_id,
77 range: 0..0,
78 with: "once ".into(),
79 },
80 }],
81 };
82 let replica = Arc::new(Replica::create("/receipts/worker.sqlite", &source).unwrap());
83 let server = Shared(Arc::new(Mutex::new(Server::new(&source))));
84 let remote = server.clone();
85 let observed = Arc::new(AtomicUsize::new(0));
86 let observations = observed.clone();
87 let worker = replica
88 .start_sync(
89 Duration::from_secs(60),
90 move || Ok(remote.clone()),
91 move |result| {
92 assert!(result.is_ok());
93 observations.fetch_add(1, Ordering::Relaxed);
94 },
95 )
96 .unwrap();
97 while observed.load(Ordering::Relaxed) == 0 {
98 next_turn().await;
99 }
100 next_turn().await;
101
102 {
103 fail_storage(1);
104 let mut applying = std::pin::pin!(replica.apply_async("Fixture", edit()));
105 assert!(matches!(
106 applying
107 .as_mut()
108 .poll(&mut Context::from_waker(Waker::noop())),
109 Poll::Pending
110 ));
111 next_turn().await;
112 fail_storage(2);
113 let mut written = std::pin::pin!(replica.written_async());
114 assert!(matches!(
115 written
116 .as_mut()
117 .poll(&mut Context::from_waker(Waker::noop())),
118 Poll::Pending
119 ));
120 next_turn().await;
121 assert!(matches!(
122 applying
123 .as_mut()
124 .poll(&mut Context::from_waker(Waker::noop())),
125 Poll::Pending
126 ));
127 assert!(matches!(
128 written
129 .as_mut()
130 .poll(&mut Context::from_waker(Waker::noop())),
131 Poll::Pending
132 ));
133 assert_eq!(replica.pending().unwrap().len(), 1);
134 assert_eq!(text(&replica.snapshot().unwrap()).2, "once original");
135 assert_eq!(server.0.lock().unwrap().publications, 0);
136 fail_storage(0);
137 phase("recovering apply and barrier");
138 notebook::fs::durable().await.unwrap();
139 applying.await.unwrap();
140 written.await.unwrap();
141 phase("publishing recovered edit");
142 while !replica.pending().unwrap().is_empty() {
143 next_turn().await;
144 }
145 }
146 assert_eq!(server.0.lock().unwrap().publications, 1);
147 assert_eq!(text(&server.0.lock().unwrap().durable).2, "once original");
148 drop(worker);
149 drop(replica);
150 next_turn().await;
151
152 let replica = Replica::create("/receipts/dropped.sqlite", &source).unwrap();
153 phase("dropping pending edit owner");
154 fail_storage(1);
155 {
156 let mut applying = std::pin::pin!(replica.apply_async("Fixture", edit()));
157 assert!(matches!(
158 applying
159 .as_mut()
160 .poll(&mut Context::from_waker(Waker::noop())),
161 Poll::Pending
162 ));
163 }
164 drop(replica);
165 match Replica::open("/receipts/dropped.sqlite") {
166 Err(error) => assert!(error.busy()),
167 Ok(_) => panic!("A pending receipt released cache ownership"),
168 }
169 next_turn().await;
170 fail_storage(0);
171 phase("reopening recovered cache");
172 let reopened = loop {
173 if let Ok(replica) = Replica::open("/receipts/dropped.sqlite") {
174 break replica;
175 }
176 next_turn().await;
177 };
178 assert_eq!(reopened.pending().unwrap().len(), 1);
179 assert_eq!(text(&reopened.snapshot().unwrap()).2, "once original");
180 drop(reopened);
181
182 phase("renaming past an evicted pending receipt");
183 let cache = std::path::Path::new("/receipts/rename-cache");
184 let page = onestore::PageCreation::new(None, Some("original"), "Fixture").unwrap();
185 let notebook = notebook::session::Notebook::create(
186 "/receipts/Before",
187 cache,
188 notebook::session::Notebook::NEW_COLOR,
189 &page,
190 )
191 .unwrap();
192 let original_cache = notebook.replica_path("New Section 1.one").unwrap();
193 let section = notebook.section("New Section 1.one", false, || {}).unwrap();
194 section.set_offline(true);
195 let (space, text_id, _) = text(&section.replica().snapshot().unwrap());
196 let replica = Arc::clone(section.replica());
197 fail_storage(1);
198 {
199 let mut applying = std::pin::pin!(replica.apply_async(
200 "Fixture",
201 Edit {
202 at,
203 ops: vec![Op::Page {
204 space,
205 op: PageOp::Text {
206 text: text_id,
207 range: 0..0,
208 with: "once ".into(),
209 },
210 }],
211 }
212 ));
213 assert!(matches!(
214 applying
215 .as_mut()
216 .poll(&mut Context::from_waker(Waker::noop())),
217 Poll::Pending
218 ));
219 }
220 section.close().unwrap();
221 drop(replica);
222 next_turn().await;
223 match notebook.rename_folder("Refused") {
224 Err(error) => assert!(error.busy()),
225 Ok(_) => panic!("A notebook moved before its replica closed"),
226 }
227 assert!(notebook::fs::metadata("/receipts/Before").is_ok());
228 assert!(notebook::fs::metadata("/receipts/Refused").is_err());
229 let notebook = notebook::session::Notebook::open("/receipts/Before", cache).unwrap();
230 let mut renaming = std::pin::pin!(notebook.rename_folder_async("After"));
231 assert!(matches!(
232 renaming
233 .as_mut()
234 .poll(&mut Context::from_waker(Waker::noop())),
235 Poll::Pending
236 ));
237 next_turn().await;
238 assert!(notebook::fs::metadata("/receipts/Before").is_ok());
239 assert!(notebook::fs::metadata("/receipts/After").is_err());
240 fail_storage(0);
241 renaming.await.unwrap();
242 let notebook = notebook::session::Notebook::open("/receipts/After", cache).unwrap();
243 let moved_cache = notebook.replica_path("New Section 1.one").unwrap();
244 assert_eq!(original_cache.file_name(), moved_cache.file_name());
245 assert!(notebook::fs::metadata(original_cache).is_err());
246 let replica = Replica::open(moved_cache).unwrap();
247 assert_eq!(replica.pending().unwrap().len(), 1);
248 assert_eq!(text(&replica.snapshot().unwrap()).2, "once original");
249 }
250}