1#![cfg(feature = "live")]
2
3use notebook::{
4 Replica,
5 live::share::{HostedRemote, Sharing},
6};
7use onestore::op::{Edit, Op, PageOp};
8use std::sync::{Arc, Barrier};
9
10#[path = "support/live.rs"]
11mod live;
12use live::*;
13
14#[test]
15fn edit_receipts_survive_guest_and_host_process_death() {
16 use notebook::live::share::{self, Guest, Host};
17 use notebook::session::Notebook;
18 use notebook::{EditStatus, PendingEdit, Remote};
19 use onestore::{CommitError, CommitState, ExGuid, Stamp, Transaction};
20 use std::{
21 collections::BTreeMap,
22 io,
23 path::{Path, PathBuf},
24 process::{Child, Command, Stdio},
25 time::{Duration, Instant},
26 };
27
28 const TEST: &str = "edit_receipts_survive_guest_and_host_process_death";
29 const ROLE: &str = "SNOWBOUND_LIVE_RECOVERY_ROLE";
30 const ROOT: &str = "SNOWBOUND_LIVE_RECOVERY_ROOT";
31 const URL: &str = "SNOWBOUND_LIVE_RECOVERY_URL";
32 const BOUNDARY: &str = "SNOWBOUND_LIVE_RECOVERY_BOUNDARY";
33 const INSERTION: &str = " CRASH_RECOVERY";
34
35 fn body(image: &[u8]) -> (ExGuid, ExGuid, String) {
36 let store = onestore::Store::parse(image).unwrap();
37 assert!(store.checksum_mismatches.is_empty());
38 let index = onestore::RevisionIndex::parse(&store).unwrap();
39 index.validate_current().unwrap();
40 let document = onestore::document::Document::parse(&index).unwrap();
41 let space = document.pages().unwrap()[0].0;
42 let (text, content) = document.spaces[&space]
43 .active()
44 .unwrap()
45 .nodes
46 .iter()
47 .find_map(|(id, node)| match &node.kind {
48 onestore::document::Kind::RichText { text, .. } => Some((*id, text.clone())),
49 _ => None,
50 })
51 .unwrap();
52 (space, text, content)
53 }
54
55 fn stopped_at_boundary(root: &Path) -> ! {
56 std::fs::write(root.join("guest-ready"), b"").unwrap();
57 loop {
58 std::thread::park();
59 }
60 }
61
62 struct Interrupted<'a> {
63 remote: HostedRemote,
64 boundary: &'a str,
65 root: &'a Path,
66 }
67 impl Remote for Interrupted<'_> {
68 fn read(&mut self) -> io::Result<Vec<u8>> {
69 self.remote.read()
70 }
71 fn stamp(&mut self) -> io::Result<Stamp> {
72 self.remote.stamp()
73 }
74 fn publish(&mut self, transaction: &Transaction) -> Result<(), CommitError> {
75 self.remote.publish(transaction)
76 }
77 fn confirm(&mut self, stamp: &Stamp) -> Result<(), CommitError> {
78 self.remote.confirm(stamp)
79 }
80 fn accepts_edits(&self) -> bool {
81 true
82 }
83 fn publish_edits(
84 &mut self,
85 transaction: &Transaction,
86 edits: &[PendingEdit],
87 revisions: &BTreeMap<ExGuid, ExGuid>,
88 ) -> Result<(), CommitError> {
89 if self.boundary == "unsent" {
90 stopped_at_boundary(self.root);
91 }
92 self.remote.publish_edits(transaction, edits, revisions)?;
93 if self.boundary == "lost" {
94 return Err(CommitError {
95 state: CommitState::Unknown,
96 error: io::ErrorKind::BrokenPipe.into(),
97 });
98 }
99 Ok(())
100 }
101 }
102
103 if let Ok(role) = std::env::var(ROLE) {
104 let root = PathBuf::from(std::env::var_os(ROOT).unwrap());
105 let url = std::env::var(URL).unwrap();
106 let boundary = std::env::var(BOUNDARY).unwrap();
107 if role == "host" {
108 let saved = root.join("sharing.json");
109 let sharing = if saved.exists() {
110 serde_json::from_slice(&std::fs::read(&saved).unwrap()).unwrap()
111 } else {
112 Sharing::new("").unwrap()
113 };
114 let host = Host::start(
115 Notebook::open(root.join("Garden"), root.join("host"))
116 .unwrap()
117 .into_storage(),
118 hello("Ada"),
119 sharing,
120 "Garden",
121 None,
122 Some(&url),
123 || {},
124 move |sharing| std::fs::write(&saved, serde_json::to_vec(sharing)?),
125 )
126 .unwrap();
127 std::fs::write(root.join("code"), code(&host)).unwrap();
128 std::fs::write(root.join("host-ready"), b"").unwrap();
129 loop {
130 std::thread::park();
131 }
132 }
133 let credentials = root.join("guest.json");
134 let (share, secret): ([u8; 16], [u8; 16]) = if role == "guest" {
135 let code = std::fs::read_to_string(root.join("code")).unwrap();
136 let welcome = share::join(hello("Alice"), &code, "", None, Some(&url)).unwrap();
137 let issued = (welcome.share, welcome.secret);
138 std::fs::write(&credentials, serde_json::to_vec(&issued).unwrap()).unwrap();
139 issued
140 } else {
141 serde_json::from_slice(&std::fs::read(credentials).unwrap()).unwrap()
142 };
143 let guest = Guest::start(hello("Alice"), share, secret, None, Some(&url), || {}).unwrap();
144 until("the restarted guest never met its host", || {
145 guest.host().is_some()
146 });
147 let database = root.join("replica.sqlite");
148 let replica = if role == "guest" {
149 let image = std::fs::read(root.join("Garden/Garden.one")).unwrap();
150 let (space, text, content) = body(&image);
151 let replica = Replica::create(&database, &image).unwrap();
152 let end = u32::try_from(content.encode_utf16().count()).unwrap();
153 let id = replica
154 .apply(
155 "Alice",
156 Edit {
157 at: 134_000_000_000_000_000,
158 ops: vec![Op::Page {
159 space,
160 op: PageOp::Text {
161 text,
162 range: end..end,
163 with: INSERTION.into(),
164 },
165 }],
166 },
167 )
168 .unwrap();
169 std::fs::write(root.join("edit-id"), id.to_string()).unwrap();
170 replica
171 } else {
172 Replica::open(&database).unwrap()
173 };
174 let id = std::fs::read_to_string(root.join("edit-id"))
175 .unwrap()
176 .parse()
177 .unwrap();
178 if role == "guest" {
179 if boundary == "before" {
180 stopped_at_boundary(&root);
181 }
182 let mut remote = Interrupted {
183 remote: HostedRemote::new(&guest, "Garden.one"),
184 boundary: &boundary,
185 root: &root,
186 };
187 let result = replica.sync_once(&mut remote);
188 if boundary == "lost" {
189 assert!(matches!(
190 result,
191 Err(notebook::Error::Remote(CommitError {
192 state: CommitState::Unknown,
193 ..
194 }))
195 ));
196 assert!(matches!(
197 replica.status(id).unwrap(),
198 Some(EditStatus::AwaitingConfirmation { .. })
199 ));
200 } else {
201 result.unwrap();
202 assert!(matches!(
203 replica.status(id).unwrap(),
204 Some(EditStatus::Published { .. })
205 ));
206 }
207 stopped_at_boundary(&root);
208 }
209 let mut remote = HostedRemote::new(&guest, "Garden.one");
210 if boundary == "unsent" {
211 let attempted = replica.status(id).unwrap();
212 assert!(matches!(
213 attempted,
214 Some(EditStatus::AwaitingConfirmation { .. })
215 ));
216 let pending = replica.pending().unwrap();
217 let snapshot = replica.snapshot().unwrap();
218 let original = std::fs::read(root.join("Garden/Garden.one")).unwrap();
219 for _ in 0..3 {
220 replica.sync_once(&mut remote).unwrap();
221 assert_eq!(replica.status(id).unwrap(), attempted);
222 assert_eq!(replica.pending().unwrap(), pending);
223 assert_eq!(replica.snapshot().unwrap(), snapshot);
224 assert_eq!(
225 std::fs::read(root.join("Garden/Garden.one")).unwrap(),
226 original
227 );
228 }
229 let archive = root.join("before-review.sqlite");
230 replica
231 .release(id, &archive, notebook::Resolution::Mine)
232 .unwrap();
233 let archive = notebook::Recovery::open(archive).unwrap();
234 assert_eq!(archive.status(id).unwrap(), attempted);
235 assert_eq!(archive.pending().unwrap(), pending);
236 assert_eq!(archive.snapshot().unwrap(), snapshot);
237 }
238 for _ in 0..5 {
239 replica.sync_once(&mut remote).unwrap();
240 if matches!(
241 replica.status(id).unwrap(),
242 Some(EditStatus::Published { .. })
243 ) {
244 break;
245 }
246 }
247 assert!(matches!(
248 replica.status(id).unwrap(),
249 Some(EditStatus::Published { .. })
250 ));
251 assert!(replica.pending().unwrap().is_empty());
252 let settled = std::fs::read(root.join("Garden/Garden.one")).unwrap();
253 for _ in 0..3 {
254 replica.sync_once(&mut remote).unwrap();
255 assert_eq!(
256 std::fs::read(root.join("Garden/Garden.one")).unwrap(),
257 settled
258 );
259 }
260 assert_eq!(replica.snapshot().unwrap(), settled);
261 drop(replica);
262 std::process::exit(0);
263 }
264
265 struct Process {
266 child: Child,
267 log: PathBuf,
268 }
269 impl Process {
270 fn start(root: &Path, url: &str, role: &str, boundary: &str) -> Self {
271 let log = root.join(format!("{role}.log"));
272 let output = std::fs::OpenOptions::new()
273 .create(true)
274 .append(true)
275 .open(&log)
276 .unwrap();
277 let child = Command::new(std::env::current_exe().unwrap())
278 .args(["--exact", TEST, "--nocapture"])
279 .env(ROLE, role)
280 .env(ROOT, root)
281 .env(URL, url)
282 .env(BOUNDARY, boundary)
283 .stdout(output.try_clone().unwrap())
284 .stderr(Stdio::from(output))
285 .spawn()
286 .unwrap();
287 Self { child, log }
288 }
289
290 fn ready(&mut self, marker: &Path) {
291 let deadline = Instant::now() + Duration::from_secs(30);
292 while !marker.exists() {
293 assert!(
294 self.child.try_wait().unwrap().is_none() && Instant::now() < deadline,
295 "{}: {}",
296 marker.display(),
297 std::fs::read_to_string(&self.log).unwrap()
298 );
299 std::thread::sleep(Duration::from_millis(20));
300 }
301 }
302
303 fn kill(&mut self) {
304 self.child.kill().unwrap();
305 assert!(!self.child.wait().unwrap().success());
306 }
307 }
308 impl Drop for Process {
309 fn drop(&mut self) {
310 let _ = self.child.kill();
311 let _ = self.child.wait();
312 }
313 }
314
315 for boundary in ["before", "unsent", "lost", "confirmed"] {
316 let directory = tempfile::tempdir().unwrap();
317 let root = directory.path();
318 let folder = notebook(root);
319 let file = folder.join("Garden.one");
320 let original =
321 include_bytes!("../../../corpus/conflict-page/native/conflict/notebook/synthetic.one");
322 std::fs::write(&file, original).unwrap();
323 let original_conflicts = server::conflicts(original);
324 assert!(!original_conflicts.is_empty());
325 let (space, _, content) = body(original);
326 let sealed = std::fs::read(folder.join("Sealed.one")).unwrap();
327 let url = relay(Default::default());
328 let mut host = Process::start(root, &url, "host", boundary);
329 host.ready(&root.join("host-ready"));
330 let mut guest = Process::start(root, &url, "guest", boundary);
331 guest.ready(&root.join("guest-ready"));
332 guest.kill();
333 host.kill();
334
335 let id: u64 = std::fs::read_to_string(root.join("edit-id"))
336 .unwrap()
337 .parse()
338 .unwrap();
339 let database = root.join("replica.sqlite");
340 let replica = Replica::open(&database).unwrap();
341 assert_eq!(
342 body(&replica.snapshot().unwrap()).2,
343 format!("{content}{INSERTION}")
344 );
345 if boundary == "confirmed" {
346 assert!(matches!(
347 replica.status(id).unwrap(),
348 Some(EditStatus::Published { .. })
349 ));
350 assert!(replica.pending().unwrap().is_empty());
351 } else if matches!(boundary, "lost" | "unsent") {
352 assert!(matches!(
353 replica.status(id).unwrap(),
354 Some(EditStatus::AwaitingConfirmation { .. })
355 ));
356 } else {
357 assert_eq!(replica.status(id).unwrap(), Some(EditStatus::Pending));
358 }
359 if boundary != "confirmed" {
360 assert_eq!(replica.pending().unwrap()[0].id, id);
361 assert_eq!(replica.pending().unwrap().len(), 1);
362 }
363 drop(replica);
364 let committed = std::fs::read(&file).unwrap();
365 if matches!(boundary, "before" | "unsent") {
366 assert_eq!(committed, original);
367 } else {
368 assert_eq!(body(&committed).2, format!("{content}{INSERTION}"));
369 }
370 let sharing: Sharing =
371 serde_json::from_slice(&std::fs::read(root.join("sharing.json")).unwrap()).unwrap();
372 let credentials = std::fs::read(root.join("guest.json")).unwrap();
373 std::fs::remove_file(root.join("host-ready")).unwrap();
374 let mut host = Process::start(root, &url, "host", boundary);
375 host.ready(&root.join("host-ready"));
376 let mut resumed = Process::start(root, &url, "resume", boundary);
377 let deadline = Instant::now() + Duration::from_secs(30);
378 let status = loop {
379 if let Some(status) = resumed.child.try_wait().unwrap() {
380 break status;
381 }
382 assert!(Instant::now() < deadline, "resumed guest timed out");
383 std::thread::sleep(Duration::from_millis(20));
384 };
385 if !status.success() {
386 let log = std::fs::read_to_string(&resumed.log).unwrap();
387 host.kill();
388 let kept = directory.keep();
389 panic!("{}: {log}", kept.display());
390 }
391 host.kill();
392 let result = std::fs::read(&file).unwrap();
393 assert_eq!(body(&result).2, format!("{content}{INSERTION}"));
394 if matches!(boundary, "lost" | "confirmed") {
395 assert_eq!(
396 &result[1024..],
397 &committed[1024..],
398 "settlement must not append another revision"
399 );
400 }
401 let revisions = |image: &[u8]| {
402 let store = onestore::Store::parse(image).unwrap();
403 onestore::RevisionIndex::parse(&store).unwrap().spaces[&space]
404 .revisions
405 .len()
406 };
407 assert_eq!(revisions(&result), revisions(original) + 1);
408 assert_eq!(server::conflicts(&result), original_conflicts);
409 assert_eq!(std::fs::read(folder.join("Sealed.one")).unwrap(), sealed);
410 let restarted: Sharing =
411 serde_json::from_slice(&std::fs::read(root.join("sharing.json")).unwrap()).unwrap();
412 assert_eq!(restarted.share, sharing.share);
413 assert_eq!(restarted.secret, sharing.secret);
414 assert_eq!(restarted.members, sharing.members);
415 assert_eq!(std::fs::read(root.join("guest.json")).unwrap(), credentials);
416 let replica = Replica::open(&database).unwrap();
417 assert!(matches!(
418 replica.status(id).unwrap(),
419 Some(EditStatus::Published { .. })
420 ));
421 assert!(replica.pending().unwrap().is_empty());
422 assert_eq!(replica.snapshot().unwrap(), result);
423 drop(replica);
424 if boundary == "unsent" {
425 let archive = notebook::Recovery::open(root.join("before-review.sqlite")).unwrap();
426 assert!(matches!(
427 archive.status(id).unwrap(),
428 Some(EditStatus::AwaitingConfirmation { .. })
429 ));
430 assert_eq!(archive.pending().unwrap()[0].id, id);
431 assert_eq!(archive.pending().unwrap().len(), 1);
432 assert_eq!(
433 body(&archive.snapshot().unwrap()).2,
434 format!("{content}{INSERTION}")
435 );
436 let database = rusqlite::Connection::open(root.join("before-review.sqlite")).unwrap();
437 assert_eq!(
438 database
439 .query_row("PRAGMA integrity_check", [], |row| row.get::<_, String>(0))
440 .unwrap(),
441 "ok"
442 );
443 println!(
444 "unsent: cold uncertainty retained; explicit Mine review archived the original queue before publication"
445 );
446 }
447 let database = rusqlite::Connection::open(&database).unwrap();
448 assert_eq!(
449 database
450 .query_row("PRAGMA integrity_check", [], |row| row.get::<_, String>(0))
451 .unwrap(),
452 "ok"
453 );
454 println!(
455 "{boundary}: original edit {id} published once, both processes restarted, SQLite integrity ok"
456 );
457 }
458}
459
460#[test]
461fn concurrent_batches_keep_receipts_and_rebase_later_keystrokes() {
462 let directory = tempfile::tempdir().unwrap();
463 let folder = notebook(directory.path());
464 let url = relay(Default::default());
465 let sharing = Sharing::new("").unwrap();
466 let host = host(&folder, &directory.path().join("host"), &sharing, &url);
467 let code = code(&host);
468 let (alice, _) = guest("Alice", &code, &url, &directory.path().join("alice"));
469 let (bob, _) = guest("Bob", &code, &url, &directory.path().join("bob"));
470 let file = folder.join("Garden.one");
471 let original = std::fs::read(&file).unwrap();
472 let (space, text, _) = server::text(&original);
473 let edit = |with: &str, at| Edit {
474 at: 134_000_000_000_000_000,
475 ops: vec![Op::Page {
476 space,
477 op: PageOp::Text {
478 text,
479 range: at..at,
480 with: with.into(),
481 },
482 }],
483 };
484 let a = Arc::new(
485 Replica::open_or_create(directory.path().join("a.sqlite"), None, || {
486 Ok(original.clone())
487 })
488 .unwrap(),
489 );
490 let b = Arc::new(
491 Replica::open_or_create(directory.path().join("b.sqlite"), None, || {
492 Ok(original.clone())
493 })
494 .unwrap(),
495 );
496 let first = a.apply("Alice", edit("A", 13)).unwrap();
497 let second = b.apply("Bob", edit("B", 0)).unwrap();
498 let barrier = Arc::new(Barrier::new(2));
499 std::thread::scope(|scope| {
500 for (replica, guest) in [(&a, &alice), (&b, &bob)] {
501 let barrier = Arc::clone(&barrier);
502 scope.spawn(move || {
503 barrier.wait();
504 replica
505 .sync_once(&mut HostedRemote::new(guest, "Garden.one"))
506 .unwrap();
507 });
508 }
509 });
510 assert!(matches!(
511 a.status(first).unwrap(),
512 Some(notebook::EditStatus::Published { .. })
513 ));
514 assert!(matches!(
515 b.status(second).unwrap(),
516 Some(notebook::EditStatus::Published { .. })
517 ));
518 let next = server::text(&a.snapshot().unwrap()).2.find('A').unwrap() as u32 + 1;
519 let third = a.apply("Alice", edit("a", next)).unwrap();
520 let mut remote = HostedRemote::new(&alice, "Garden.one");
521 for _ in 0..5 {
522 a.sync_once(&mut remote).unwrap();
523 if matches!(
524 a.status(third).unwrap(),
525 Some(notebook::EditStatus::Published { .. })
526 ) {
527 break;
528 }
529 }
530 let result = std::fs::read(&file).unwrap();
531 let result_text = server::text(&result).2;
532 assert!(result_text.contains("Aa"), "{result_text}");
533 assert!(result_text.contains('B'), "{result_text}");
534 let arena = onestore::Arena::default();
535 let mut section = onestore::Section::open(&arena, result).unwrap();
536 assert!(section.conflicts().unwrap().is_empty());
537}