1//! A fake document provider with real descriptors: seekable files, or pipes as the Android lab
2//! measured stream-only providers (`tools/android/provider-lab`).
3
4use super::*;
5use crate::{
6 EditStatus, Replica,
7 session::{self, Background, Event, Notebook, SyncState, SyncStatus},
8};
9use onestore::{
10 ExGuid,
11 op::{Edit, Op, PageOp},
12 page::{Page, PageObject},
13};
14use std::{
15 collections::HashMap,
16 fs,
17 ops::Range,
18 os::fd::{AsRawFd, OwnedFd},
19 path::Path,
20 thread,
21 time::{Duration, Instant},
22};
23
24const TOC: &str = "Open Notebook.onetoc2";
25
26#[derive(Clone, Copy, PartialEq)]
27enum Kind {
28 Seekable,
29 /// Pipes in every mode, `rw` included, with no metadata; `truncate` where `wt` is taken.
30 Stream {
31 truncate: bool,
32 },
33}
34
35struct Node {
36 parent: String,
37 name: String,
38 directory: bool,
39}
40
41struct State {
42 kind: Kind,
43 nodes: BTreeMap<String, Node>,
44 next: u32,
45 writable: bool,
46 offline: bool,
47 /// Lists, from its cache, but opens nothing, as a cloud provider offline does.
48 unopenable: bool,
49 revoked: bool,
50 /// One device's writes reach the others only at `publish_staged`, as the lab's
51 /// upload-on-close provider showed.
52 delayed: bool,
53 /// The next `rw` open finds its document renamed to `Moved.one` under a new id, and a
54 /// document holding these bytes in its place.
55 swap: Option<Vec<u8>>,
56 /// The next stream upload fails after this many bytes, leaving them as the document where
57 /// `.1`, as a provider that writes non-atomically does, or else the document as it was.
58 cut: Option<(u64, bool)>,
59 /// Streams refuse to open for writing, as a provider out of space does.
60 full: bool,
61 modes: Vec<String>,
62 /// Each stream being uploaded, by descriptor: its document, whether `wt` truncates it, and
63 /// what arrived with the cut that stopped it, if one did.
64 uploads: HashMap<i32, (String, bool, Upload)>,
65}
66
67type Upload = thread::JoinHandle<(Vec<u8>, Option<(u64, bool)>)>;
68
69struct Shared {
70 folder: tempfile::TempDir,
71 state: Mutex<State>,
72}
73
74/// One device's view of the provider.
75#[derive(Clone)]
76struct Fake {
77 store: Arc<Shared>,
78 device: u32,
79}
80
81impl Fake {
82 fn new(kind: Kind) -> Self {
83 let mut nodes = BTreeMap::new();
84 nodes.insert(
85 "root".to_owned(),
86 Node {
87 parent: String::new(),
88 name: String::new(),
89 directory: true,
90 },
91 );
92 let state = State {
93 kind,
94 nodes,
95 next: 0,
96 writable: true,
97 offline: false,
98 unopenable: false,
99 revoked: false,
100 delayed: false,
101 swap: None,
102 cut: None,
103 full: false,
104 modes: Vec::new(),
105 uploads: HashMap::new(),
106 };
107 Self {
108 store: Arc::new(Shared {
109 folder: tempfile::tempdir().unwrap(),
110 state: Mutex::new(state),
111 }),
112 device: 0,
113 }
114 }
115
116 fn device(&self, device: u32) -> Self {
117 Self {
118 store: Arc::clone(&self.store),
119 device,
120 }
121 }
122
123 fn documents(&self, root: &str) -> Arc<Documents> {
124 let cache = self
125 .store
126 .folder
127 .path()
128 .join(format!("cache-{}", self.device));
129 let device = format!("Device {}", self.device);
130 Arc::new(Documents::new(self.clone(), root, cache, device))
131 }
132
133 fn state(&self) -> MutexGuard<'_, State> {
134 lock(&self.store.state)
135 }
136
137 fn file(&self, id: &str) -> PathBuf {
138 self.store.folder.path().join(id)
139 }
140
141 fn staged(&self, id: &str) -> PathBuf {
142 self.store
143 .folder
144 .path()
145 .join(format!("{id}.{}", self.device))
146 }
147
148 /// The bytes this device reads.
149 fn visible(&self, id: &str) -> PathBuf {
150 let staged = self.staged(id);
151 if staged.exists() {
152 staged
153 } else {
154 self.file(id)
155 }
156 }
157
158 fn add(&self, name: &str, bytes: &[u8]) -> String {
159 let mut state = self.state();
160 state.next += 1;
161 let id = format!("doc-{:04}", state.next);
162 state.nodes.insert(
163 id.clone(),
164 Node {
165 parent: "root".into(),
166 name: name.into(),
167 directory: false,
168 },
169 );
170 fs::write(self.file(&id), bytes).unwrap();
171 id
172 }
173
174 /// The first document named `name` at the root, as stored.
175 fn id(&self, name: &str) -> String {
176 let state = self.state();
177 let (id, _) = state
178 .nodes
179 .iter()
180 .find(|(_, node)| node.parent == "root" && node.name == name)
181 .unwrap();
182 id.clone()
183 }
184
185 fn bytes(&self, name: &str) -> Vec<u8> {
186 fs::read(self.file(&self.id(name))).unwrap()
187 }
188
189 fn set(&self, change: impl FnOnce(&mut State)) {
190 change(&mut self.state());
191 }
192
193 fn modes(&self) -> Vec<String> {
194 self.state().modes.clone()
195 }
196
197 /// Makes every device's writes the document, in device order, as an explicit provider
198 /// publication does.
199 fn publish_staged(&self) {
200 let ids: Vec<String> = self.state().nodes.keys().cloned().collect();
201 for id in ids {
202 for device in 0..4 {
203 let staged = self.device(device).staged(&id);
204 if staged.exists() {
205 fs::rename(staged, self.file(&id)).unwrap();
206 }
207 }
208 }
209 }
210
211 /// Gives the document a new id, as another app renaming it does; the old one is gone.
212 fn reidentify(&self, name: &str) -> String {
213 let old = self.id(name);
214 let mut state = self.state();
215 state.next += 1;
216 let new = format!("doc-{:04}", state.next);
217 let node = state.nodes.remove(&old).unwrap();
218 state.nodes.insert(new.clone(), node);
219 fs::rename(self.file(&old), self.file(&new)).unwrap();
220 new
221 }
222
223 fn reachable(&self) -> io::Result<MutexGuard<'_, State>> {
224 let state = self.state();
225 if state.offline {
226 return Err(io::Error::new(io::ErrorKind::NotConnected, "Offline"));
227 }
228 if state.revoked {
229 return Err(io::Error::new(io::ErrorKind::PermissionDenied, "Revoked"));
230 }
231 Ok(state)
232 }
233
234 fn describe(&self, state: &State, id: &str) -> io::Result<Document> {
235 let node = state.nodes.get(id).ok_or(io::ErrorKind::NotFound)?;
236 let metadata = fs::metadata(self.visible(id)).ok();
237 let known = |value: u64| (state.kind == Kind::Seekable).then_some(value);
238 Ok(Document {
239 id: id.to_owned(),
240 name: node.name.clone(),
241 directory: node.directory,
242 size: metadata.as_ref().and_then(|metadata| known(metadata.len())),
243 modified: metadata
244 .and_then(|metadata| metadata.modified().ok())
245 .and_then(|at| at.duration_since(std::time::UNIX_EPOCH).ok())
246 .and_then(|since| known(since.as_nanos() as u64)),
247 writable: state.writable,
248 })
249 }
250}
251
252impl Provider for Fake {
253 fn document(&self, id: &str) -> io::Result<Document> {
254 let state = self.reachable()?;
255 self.describe(&state, id)
256 }
257
258 fn children(&self, id: &str) -> io::Result<Vec<Document>> {
259 let state = self.reachable()?;
260 if !state
261 .nodes
262 .get(id)
263 .ok_or(io::ErrorKind::NotFound)?
264 .directory
265 {
266 return Err(io::ErrorKind::NotADirectory.into());
267 }
268 let children = state.nodes.iter().filter(|(_, node)| node.parent == id);
269 children.map(|(id, _)| self.describe(&state, id)).collect()
270 }
271
272 fn open(&self, id: &str, mode: &str) -> io::Result<File> {
273 let mut state = self.reachable()?;
274 if state.unopenable {
275 return Err(io::Error::new(io::ErrorKind::NotConnected, "Offline"));
276 }
277 if !state.nodes.contains_key(id) {
278 return Err(io::ErrorKind::NotFound.into());
279 }
280 state.modes.push(mode.to_owned());
281 if mode != "r" && !state.writable {
282 return Err(io::Error::new(
283 io::ErrorKind::PermissionDenied,
284 "Read-only grant",
285 ));
286 }
287 if mode == "rw"
288 && let Some(twin) = state.swap.take()
289 {
290 let mut node = state.nodes.remove(id).unwrap();
291 let name = std::mem::replace(&mut node.name, "Moved.one".into());
292 let parent = node.parent.clone();
293 let (moved, taken) = (format!("{id}-moved"), format!("{id}-taken"));
294 state.nodes.insert(moved.clone(), node);
295 let directory = false;
296 state.nodes.insert(
297 taken.clone(),
298 Node {
299 parent,
300 name,
301 directory,
302 },
303 );
304 fs::rename(self.file(id), self.file(&moved))?;
305 fs::write(self.file(&taken), twin)?;
306 return Err(io::ErrorKind::NotFound.into());
307 }
308 match (state.kind, mode) {
309 (Kind::Seekable, "r") => File::open(self.visible(id)),
310 (Kind::Seekable, _) => {
311 let mut file = self.file(id);
312 if state.delayed {
313 let staged = self.staged(id);
314 if !staged.exists() {
315 fs::copy(&file, &staged)?;
316 }
317 file = staged;
318 }
319 File::options()
320 .read(true)
321 .write(true)
322 .truncate(mode == "wt")
323 .open(file)
324 }
325 (Kind::Stream { .. }, "r") => {
326 let (reader, mut writer) = io::pipe()?;
327 let bytes = fs::read(self.file(id))?;
328 thread::spawn(move || writer.write_all(&bytes));
329 Ok(File::from(OwnedFd::from(reader)))
330 }
331 (Kind::Stream { .. }, "rw") => Ok(File::from(OwnedFd::from(io::pipe()?.1))),
332 // As Android's host reports a refused mode: unreachable, like most failures.
333 (Kind::Stream { truncate: false }, "wt") => Err(io::Error::new(
334 io::ErrorKind::NotConnected,
335 "Unsupported mode",
336 )),
337 (Kind::Stream { .. }, _) if state.full => Err(io::Error::other("No space left")),
338 (Kind::Stream { .. }, _) => {
339 let (reader, writer) = io::pipe()?;
340 let cut = state.cut.take();
341 let upload = thread::spawn(move || {
342 let mut received = Vec::new();
343 let at = cut.map_or(u64::MAX, |(at, _)| at);
344 let _ = reader.take(at).read_to_end(&mut received);
345 (received, cut)
346 });
347 let writer = File::from(OwnedFd::from(writer));
348 let upload = (id.to_owned(), mode == "wt", upload);
349 state.uploads.insert(writer.as_raw_fd(), upload);
350 Ok(writer)
351 }
352 }
353 }
354
355 fn close(&self, file: File) -> io::Result<()> {
356 let upload = self.state().uploads.remove(&file.as_raw_fd());
357 drop(file);
358 let Some((id, truncate, upload)) = upload else {
359 return Ok(());
360 };
361 let (received, cut) = upload.join().unwrap();
362 if let Some((_, torn)) = cut {
363 if torn {
364 fs::write(self.file(&id), received)?;
365 }
366 return Err(io::Error::other("The upload failed"));
367 }
368 let mut bytes = if truncate {
369 Vec::new()
370 } else {
371 fs::read(self.file(&id))?
372 };
373 if bytes.len() < received.len() {
374 bytes.resize(received.len(), 0);
375 }
376 bytes[..received.len()].copy_from_slice(&received);
377 fs::write(self.file(&id), bytes)
378 }
379
380 fn rename(&self, id: &str, name: &str) -> io::Result<String> {
381 let mut state = self.reachable()?;
382 let mut node = state.nodes.remove(id).ok_or(io::ErrorKind::NotFound)?;
383 node.name = name.to_owned();
384 state.next += 1;
385 let new = format!("doc-{:04}", state.next);
386 state.nodes.insert(new.clone(), node);
387 fs::rename(self.file(id), self.file(&new))?;
388 Ok(new)
389 }
390}
391
392/// A section holding `text`, edited once since it was made, as a section OneNote has written
393/// is: the first edit of a page `create_section` makes starts a fresh revision chain, so its
394/// pages' later states trace back to that edit, not to the section as made.
395fn section_image(name: &str, text: &str) -> Vec<u8> {
396 let (first, rest) = text.split_at(1);
397 let mut image = onestore::create_section(name, rest, "Author").unwrap();
398 let arena = onestore::Arena::default();
399 let mut section = onestore::Section::open(&arena, image.clone()).unwrap();
400 let space = section.pages().unwrap()[0].0;
401 let op = PageOp::Text {
402 text: paragraph(&section.page(space).unwrap()).0,
403 range: 0..0,
404 with: first.into(),
405 };
406 let edit = Edit {
407 at: crate::now(),
408 ops: vec![Op::Page { space, op }],
409 };
410 section.apply("Author", &edit).unwrap();
411 section.seal().unwrap().unwrap().apply(&mut image).unwrap();
412 image
413}
414
415/// A notebook folder: its TOC and two sections it lists, placed under it.
416fn notebook(fake: &Fake) {
417 let mut notes = section_image("Notes.one", "Original text");
418 let mut ideas = section_image("Ideas.one", "First idea");
419 let file = |image: &[u8]| Store::parse(image).unwrap().header.file_id;
420 let toc = onestore::create_table_of_contents(
421 TOC,
422 &[("Notes.one", file(&notes)), ("Ideas.one", file(&ideas))],
423 )
424 .unwrap();
425 onestore::place_image(&mut notes, Some((file(&toc), "Notes.one"))).unwrap();
426 onestore::place_image(&mut ideas, Some((file(&toc), "Ideas.one"))).unwrap();
427 fake.add(TOC, &toc);
428 fake.add("Notes.one", &notes);
429 fake.add("Ideas.one", &ideas);
430}
431
432/// Opens a section of `notebook` as a host does: its replica, made from the file only where
433/// there is none.
434fn open(notebook: &Notebook, documents: &Arc<Documents>, path: &str) -> session::Section {
435 let cache = notebook.replica_path(path).unwrap();
436 fs::create_dir_all(cache.parent().unwrap()).unwrap();
437 let replica = Replica::open_or_create(&cache, None, || notebook.read_section(path)).unwrap();
438 session::Section::resume_provider(path.into(), replica, Arc::clone(documents), false, || {})
439 .unwrap()
440}
441
442fn paragraph(page: &Page) -> (ExGuid, String) {
443 page.objects
444 .iter()
445 .find_map(|object| match object {
446 PageObject::Outline(outline) => outline.paragraphs.iter().find_map(|paragraph| {
447 let text = paragraph.text()?;
448 Some((text.id, text.text.text().to_owned()))
449 }),
450 _ => None,
451 })
452 .unwrap()
453}
454
455fn first_page(section: &session::Section) -> (ExGuid, Page) {
456 let space = section.pages().unwrap()[0].0;
457 (space, section.page(space).unwrap())
458}
459
460fn text(section: &session::Section) -> String {
461 paragraph(&first_page(section).1).1
462}
463
464/// Replaces `range` of the first page's text as the editor does, returning the edit's id.
465fn typed(section: &session::Section, range: Range<u32>, with: &str) -> u64 {
466 let (space, page) = first_page(section);
467 let op = PageOp::Text {
468 text: paragraph(&page).0,
469 range,
470 with: with.into(),
471 };
472 let edit = Edit {
473 at: crate::now(),
474 ops: vec![Op::Page { space, op }],
475 };
476 section.replica().apply("Editor", edit).unwrap()
477}
478
479/// The first page's text in `image`, which must cold-open as a section.
480fn stored(image: Vec<u8>) -> String {
481 let arena = onestore::Arena::default();
482 let mut section = onestore::Section::open(&arena, image).unwrap();
483 let space = section.pages().unwrap()[0].0;
484 paragraph(&section.page(space).unwrap()).1
485}
486
487fn until(what: &str, mut done: impl FnMut() -> bool) {
488 let deadline = Instant::now() + Duration::from_secs(20);
489 while !done() {
490 assert!(Instant::now() < deadline, "{what}");
491 thread::sleep(Duration::from_millis(20));
492 }
493}
494
495fn published(section: &session::Section, id: u64) {
496 until("the edit was published", || {
497 section.wake();
498 matches!(
499 section.status(id).unwrap(),
500 Some(EditStatus::Published { .. })
501 )
502 });
503}
504
505fn status(section: &session::Section, accept: impl Fn(&SyncStatus) -> bool) -> SyncStatus {
506 let mut last = None;
507 until("the status came", || {
508 section.wake();
509 let status = section.sync_status().unwrap();
510 let accepted = accept(&status);
511 last = Some(status);
512 accepted
513 });
514 last.unwrap()
515}
516
517#[test]
518fn seekable_documents_take_commits_in_place_and_rebase_on_another_devices() {
519 let fake = Fake::new(Kind::Seekable);
520 notebook(&fake);
521 let documents = fake.documents("root");
522 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
523 assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]);
524 let section = open(&notebook, &documents, "Notes.one");
525 let before = fake.bytes("Notes.one").len();
526 let id = typed(&section, 0..0, "Hello ");
527 published(&section, id);
528 let after = fake.bytes("Notes.one");
529 assert!(after.len() > before, "the revision was appended");
530 assert_eq!(stored(after), "Hello Original text");
531 assert!(fake.modes().iter().all(|mode| mode == "r" || mode == "rw"));
532
533 let elsewhere = fake.device(1).documents("root");
534 let other = Notebook::open_provider(Arc::clone(&elsewhere)).unwrap();
535 let other = open(&other, &elsewhere, "Notes.one");
536 let id = typed(&other, 0..0, "World ");
537 published(&other, id);
538 until("the change reached the first device", || {
539 section.wake();
540 text(&section) == "World Hello Original text"
541 });
542 assert!(
543 section
544 .events()
545 .iter()
546 .any(|event| matches!(event, Event::Changed(_)))
547 );
548 assert_eq!(stored(fake.bytes("Notes.one")), "World Hello Original text");
549 section.close().unwrap();
550 other.close().unwrap();
551}
552
553#[test]
554fn stream_only_documents_take_whole_images_and_never_open_to_write_in_place() {
555 for truncate in [true, false] {
556 let fake = Fake::new(Kind::Stream { truncate });
557 notebook(&fake);
558 let documents = fake.documents("root");
559 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
560 let section = open(&notebook, &documents, "Ideas.one");
561 let id = typed(&section, 0..0, "Second. ");
562 published(&section, id);
563 let id = typed(&section, 0..0, "Third. ");
564 published(&section, id);
565 assert_eq!(stored(fake.bytes("Ideas.one")), "Third. Second. First idea");
566 let modes = fake.modes();
567 let count = |wanted: &str| modes.iter().filter(|mode| *mode == wanted).count();
568 assert_eq!(count("rw"), 0, "{modes:?}");
569 assert_eq!((count("wt"), count("w")), (2, if truncate { 0 } else { 2 }));
570 section.close().unwrap();
571 }
572}
573
574#[test]
575fn documents_without_a_write_grant_are_only_read() {
576 let fake = Fake::new(Kind::Seekable);
577 notebook(&fake);
578 fake.set(|state| state.writable = false);
579 let documents = fake.documents("root");
580 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
581 let section = open(&notebook, &documents, "Notes.one");
582 let before = fake.bytes("Notes.one");
583 typed(&section, 0..0, "Mine ");
584 let status = status(&section, |status| status.error.is_some());
585 assert_eq!(status.state(), SyncState::ReadOnly);
586 assert_eq!(
587 status.error.unwrap().kind(),
588 io::ErrorKind::ReadOnlyFilesystem
589 );
590 assert_eq!(status.queued, 1);
591 assert_eq!(fake.bytes("Notes.one"), before);
592 assert!(fake.modes().iter().all(|mode| mode == "r"));
593 section.close().unwrap();
594}
595
596#[test]
597fn a_write_another_device_sees_late_rebases_it_once_visible() {
598 let fake = Fake::new(Kind::Seekable);
599 notebook(&fake);
600 fake.set(|state| state.delayed = true);
601 let (writer, reader) = (fake.documents("root"), fake.device(1).documents("root"));
602 let writing = Notebook::open_provider(Arc::clone(&writer)).unwrap();
603 let reading = Notebook::open_provider(Arc::clone(&reader)).unwrap();
604 let (writing, reading) = (
605 open(&writing, &writer, "Notes.one"),
606 open(&reading, &reader, "Notes.one"),
607 );
608 let id = typed(&writing, 0..0, "Early ");
609 published(&writing, id);
610 // The writer's close succeeded; the other device still reads the old bytes.
611 let old = reader.read("Notes.one", crate::MAX_FILE_BYTES).unwrap();
612 assert_eq!(stored(old), "Original text");
613 reading.wake();
614 thread::sleep(Duration::from_millis(200));
615 assert_eq!(text(&reading), "Original text");
616 reading.set_offline(true);
617 let id = typed(&reading, 13..13, " later");
618 fake.publish_staged();
619 reading.set_offline(false);
620 published(&reading, id);
621 assert_eq!(text(&reading), "Early Original text later");
622 fake.publish_staged();
623 assert_eq!(stored(fake.bytes("Notes.one")), "Early Original text later");
624 writing.close().unwrap();
625 reading.close().unwrap();
626}
627
628#[test]
629fn a_cold_start_while_offline_opens_the_last_listing_and_queued_edits() {
630 let fake = Fake::new(Kind::Seekable);
631 notebook(&fake);
632 let documents = fake.documents("root");
633 let mut notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
634 let background = Background::provider(Arc::clone(&documents), || {}).unwrap();
635 background.watch(notebook.replicas());
636 until("every section has its offline copy", || {
637 ["Notes.one", "Ideas.one"]
638 .iter()
639 .all(|path| notebook.replica_path(path).unwrap().exists())
640 });
641 background.stop();
642 let section = open(&notebook, &documents, "Ideas.one");
643 fake.set(|state| state.offline = true);
644 let id = typed(&section, 0..0, "Offline ");
645 status(&section, |status| {
646 status.state() == SyncState::NotConnected && status.queued == 1
647 });
648 section.close().unwrap();
649 drop(notebook);
650
651 // A new process: the provider fails throughout.
652 let documents = fake.documents("root");
653 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
654 assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]);
655 let notes = open(&notebook, &documents, "Notes.one");
656 assert_eq!(text(&notes), "Original text");
657 let section = open(&notebook, &documents, "Ideas.one");
658 assert_eq!(text(&section), "Offline First idea");
659 assert_eq!(section.pending().unwrap()[0].id, id);
660 status(&section, |status| status.state() == SyncState::NotConnected);
661 fake.set(|state| state.offline = false);
662 published(&section, id);
663 assert_eq!(stored(fake.bytes("Ideas.one")), "Offline First idea");
664 notes.close().unwrap();
665 section.close().unwrap();
666}
667
668#[test]
669fn a_revoked_grant_is_not_offline_and_keeps_the_replicas() {
670 let fake = Fake::new(Kind::Seekable);
671 notebook(&fake);
672 let documents = fake.documents("root");
673 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
674 let replica = notebook.replica_path("Notes.one").unwrap();
675 let section = open(&notebook, &documents, "Notes.one");
676 fake.set(|state| state.revoked = true);
677 let id = typed(&section, 0..0, "Kept ");
678 let status = status(&section, |status| status.error.is_some());
679 assert_eq!(
680 status.error.unwrap().kind(),
681 io::ErrorKind::PermissionDenied
682 );
683 section.close().unwrap();
684 drop(notebook);
685 let refused = Notebook::open_provider(fake.documents("root")).err();
686 assert!(
687 matches!(
688 &refused,
689 Some(Error::Discovery(discover::Error::Io { error, .. }))
690 if error.kind() == io::ErrorKind::PermissionDenied
691 ),
692 "{refused:?}"
693 );
694 assert_eq!(
695 Replica::open(&replica).unwrap().pending().unwrap()[0].id,
696 id
697 );
698
699 fake.set(|state| state.revoked = false);
700 let documents = fake.documents("root");
701 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
702 let section = open(&notebook, &documents, "Notes.one");
703 published(&section, id);
704 assert_eq!(stored(fake.bytes("Notes.one")), "Kept Original text");
705 section.close().unwrap();
706}
707
708#[test]
709fn documents_renamed_to_new_ids_are_listed_again() {
710 let fake = Fake::new(Kind::Seekable);
711 notebook(&fake);
712 let documents = fake.documents("root");
713 let mut notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
714 let section = open(&notebook, &documents, "Notes.one");
715 let id = typed(&section, 0..0, "One ");
716 published(&section, id);
717 fake.reidentify("Notes.one");
718 let id = typed(&section, 0..0, "Two ");
719 published(&section, id);
720 assert_eq!(stored(fake.bytes("Notes.one")), "Two One Original text");
721 section.close().unwrap();
722
723 let old = fake.id("Notes.one");
724 assert_eq!(
725 notebook.rename("Notes.one", "Journal").unwrap(),
726 "Journal.one"
727 );
728 assert!(matches!(
729 fake.document(&old).map_err(|error| error.kind()),
730 Err(io::ErrorKind::NotFound)
731 ));
732 assert_eq!(notebook.catalog().order, ["Journal.one", "Ideas.one"]);
733 let identity = notebook.catalog().sections[0].file_id;
734 assert!(session::lists(&fake.bytes(TOC), identity).unwrap());
735 let section = open(&notebook, &documents, "Journal.one");
736 let id = typed(&section, 0..0, "Three ");
737 published(&section, id);
738 assert_eq!(
739 stored(fake.bytes("Journal.one")),
740 "Three Two One Original text"
741 );
742 section.close().unwrap();
743}
744
745#[test]
746fn of_documents_sharing_a_name_one_lists_and_the_rest_are_unavailable() {
747 let fake = Fake::new(Kind::Seekable);
748 notebook(&fake);
749 let twin = fake.add(
750 "Notes.one",
751 &onestore::create_section("Notes.one", "A twin", "Author").unwrap(),
752 );
753 let documents = fake.documents("root");
754 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
755 assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]);
756 let unavailable = &notebook.catalog().unavailable;
757 assert_eq!(unavailable.len(), 1);
758 assert_eq!(unavailable[0].path, format!("Notes.one ({twin})"));
759 let section = open(&notebook, &documents, "Notes.one");
760 assert_eq!(text(&section), "Original text");
761 section.close().unwrap();
762}
763
764#[test]
765fn a_document_taking_the_name_between_check_and_write_is_not_written() {
766 let fake = Fake::new(Kind::Seekable);
767 notebook(&fake);
768 let documents = fake.documents("root");
769 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
770 let section = open(&notebook, &documents, "Notes.one");
771 let id = typed(&section, 0..0, "Checked ");
772 published(&section, id);
773 let original = fake.bytes("Notes.one");
774 let twin = onestore::create_section("Notes.one", "A twin", "Author").unwrap();
775 fake.set(|state| state.swap = Some(twin.clone()));
776 typed(&section, 0..0, "Refused ");
777 until("the write was refused", || {
778 section.wake();
779 fake.state().swap.is_none() && section.sync_status().unwrap().error.is_some()
780 });
781 for _ in 0..3 {
782 section.wake();
783 thread::sleep(Duration::from_millis(100));
784 }
785 assert_eq!(fake.bytes("Notes.one"), twin);
786 assert_eq!(fake.bytes("Moved.one"), original);
787 // Another section under the name stops the worker, as it does for a file.
788 assert!(section.close().is_err());
789}
790
791#[test]
792fn place_and_confirm_write_a_seekable_document_in_place() {
793 let fake = Fake::new(Kind::Seekable);
794 notebook(&fake);
795 let listed = tempfile::tempdir().unwrap();
796 let tree = Tree {
797 documents: fake.documents("root"),
798 listed: listed.path().join("entries.json"),
799 };
800 let before = fake.bytes("Ideas.one");
801 let stamp = Stamp::of(&before).unwrap();
802 tree.confirm("Ideas.one", &stamp).unwrap();
803 let confirmed = fake.bytes("Ideas.one");
804 assert_eq!(confirmed.len(), before.len());
805 assert_ne!(confirmed[212..252], before[212..252]);
806 assert_eq!(stored(confirmed.clone()), "First idea");
807 let stale = tree.confirm("Ideas.one", &stamp).unwrap_err();
808 assert_eq!(stale.state, CommitState::NotCommitted);
809 tree.place("Ideas.one", [7; 16], "Renamed.one").unwrap();
810 let mut placed = confirmed;
811 onestore::place_image(&mut placed, Some(([7; 16], "Renamed.one"))).unwrap();
812 assert_eq!(fake.bytes("Ideas.one"), placed);
813 assert!(fake.modes().iter().all(|mode| mode == "r" || mode == "rw"));
814}
815
816#[test]
817fn conflicting_edits_leave_a_conflict_page_in_the_document() {
818 let fake = Fake::new(Kind::Seekable);
819 notebook(&fake);
820 let (mine, theirs) = (fake.documents("root"), fake.device(1).documents("root"));
821 let local = Notebook::open_provider(Arc::clone(&mine)).unwrap();
822 let remote = Notebook::open_provider(Arc::clone(&theirs)).unwrap();
823 let (local, remote) = (
824 open(&local, &mine, "Notes.one"),
825 open(&remote, &theirs, "Notes.one"),
826 );
827 local.set_offline(true);
828 typed(&local, 0..8, "Local");
829 let id = typed(&remote, 0..8, "Native");
830 published(&remote, id);
831 local.set_offline(false);
832 until("the conflict page was published", || {
833 local.wake();
834 !local.conflicts().unwrap().is_empty() && local.pending().unwrap().is_empty()
835 });
836 assert_eq!(text(&local), "Native text");
837 let conflicts = local.conflicts().unwrap();
838 let kept = local.page(conflicts[0].1[0].space).unwrap();
839 assert_eq!(paragraph(&kept).1, "Local text");
840 let arena = onestore::Arena::default();
841 let mut stored = onestore::Section::open(&arena, fake.bytes("Notes.one")).unwrap();
842 assert_eq!(stored.conflicts().unwrap().len(), 1);
843 local.close().unwrap();
844 remote.close().unwrap();
845}
846
847#[test]
848fn a_lone_document_opens_offline_from_its_replica() {
849 let fake = Fake::new(Kind::Seekable);
850 let image = onestore::create_section("Lone.one", "Alone", "Author").unwrap();
851 let lone = fake.add("Lone.one", &image);
852 let open =
853 |fake: &Fake| session::Section::open_provider(fake.documents(&lone), false, || {}).unwrap();
854 let section = open(&fake);
855 let id = typed(&section, 0..0, "Not ");
856 published(&section, id);
857 section.close().unwrap();
858 fake.set(|state| state.offline = true);
859 let section = open(&fake);
860 assert_eq!(section.file(), Path::new(&lone));
861 assert_eq!(text(&section), "Not Alone");
862 let id = typed(&section, 9..9, " now");
863 status(&section, |status| status.state() == SyncState::NotConnected);
864 fake.set(|state| state.offline = false);
865 published(&section, id);
866 assert_eq!(stored(fake.bytes("Lone.one")), "Not Alone now");
867 section.close().unwrap();
868}
869
870#[test]
871fn a_provider_that_lists_but_cannot_open_opens_sections_from_their_replicas() {
872 let fake = Fake::new(Kind::Seekable);
873 notebook(&fake);
874 let documents = fake.documents("root");
875 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
876 open(&notebook, &documents, "Notes.one").close().unwrap();
877 drop(notebook);
878 // Another device changes the section, so it lists otherwise than when last read.
879 let elsewhere = fake.device(1).documents("root");
880 let other = Notebook::open_provider(Arc::clone(&elsewhere)).unwrap();
881 let other = open(&other, &elsewhere, "Notes.one");
882 let id = typed(&other, 0..0, "Theirs ");
883 published(&other, id);
884 other.close().unwrap();
885 fake.set(|state| state.unopenable = true);
886
887 let documents = fake.documents("root");
888 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
889 assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]);
890 let section = open(&notebook, &documents, "Notes.one");
891 assert_eq!(text(&section), "Original text");
892 let id = typed(&section, 13..13, " mine");
893 status(&section, |status| {
894 status.state() == SyncState::NotConnected && status.queued == 1
895 });
896 fake.set(|state| state.unopenable = false);
897 published(&section, id);
898 assert_eq!(stored(fake.bytes("Notes.one")), "Theirs Original text mine");
899 section.close().unwrap();
900}
901
902/// The images a device keeps of what it published, for every notebook in its cache.
903fn kept_images(fake: &Fake) -> Vec<String> {
904 let cache = fake
905 .store
906 .folder
907 .path()
908 .join(format!("cache-{}", fake.device));
909 let folders = fs::read_dir(cache.join("replicas")).into_iter().flatten();
910 folders
911 .filter_map(|folder| fs::read_dir(folder.ok()?.path().join("published")).ok())
912 .flatten()
913 .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
914 .filter(|name| name.ends_with(".one"))
915 .collect()
916}
917
918/// The images a device keeps that Notes.one does not stand as, still to merge.
919fn unmerged(fake: &Fake) -> Vec<String> {
920 let current = digest(&Stamp::of(&fake.bytes("Notes.one")).unwrap());
921 let mut images = kept_images(fake);
922 images.retain(|name| Kept::named(name.clone()).is_some_and(|kept| kept.image != current));
923 images
924}
925
926#[test]
927fn a_publication_another_devices_upload_replaced_merges_back() {
928 let fake = Fake::new(Kind::Seekable);
929 notebook(&fake);
930 fake.set(|state| state.delayed = true);
931 let (mine, theirs) = (fake.documents("root"), fake.device(1).documents("root"));
932 let (local, remote) = (
933 Notebook::open_provider(Arc::clone(&mine)).unwrap(),
934 Notebook::open_provider(Arc::clone(&theirs)).unwrap(),
935 );
936 let (local, remote) = (
937 open(&local, &mine, "Notes.one"),
938 open(&remote, &theirs, "Notes.one"),
939 );
940 let id = typed(&local, 0..0, "Alpha ");
941 published(&local, id);
942 let id = typed(&remote, 13..13, " beta");
943 published(&remote, id);
944 // Both uploads land; the later one, which never saw the first, is the document.
945 fake.publish_staged();
946 assert_eq!(stored(fake.bytes("Notes.one")), "Original text beta");
947 until("the lost publication merged back", || {
948 local.wake();
949 fake.publish_staged();
950 stored(fake.bytes("Notes.one")) == "Alpha Original text beta"
951 });
952 until("both devices show it", || {
953 local.wake();
954 remote.wake();
955 text(&local) == "Alpha Original text beta" && text(&remote) == text(&local)
956 });
957 assert!(local.conflicts().unwrap().is_empty());
958 assert!(unmerged(&fake).is_empty());
959 local.close().unwrap();
960 remote.close().unwrap();
961}
962
963/// Ages the images a device keeps past `LAG`, as if reads had long had time to catch up.
964fn aged(fake: &Fake) {
965 let cache = fake
966 .store
967 .folder
968 .path()
969 .join(format!("cache-{}", fake.device));
970 for folder in fs::read_dir(cache.join("replicas")).unwrap() {
971 for image in fs::read_dir(folder.unwrap().path().join("published"))
972 .into_iter()
973 .flatten()
974 {
975 let file = File::options()
976 .write(true)
977 .open(image.unwrap().path())
978 .unwrap();
979 let past = std::time::SystemTime::now() - LAG - Duration::from_secs(1);
980 file.set_modified(past).unwrap();
981 }
982 }
983}
984
985fn failed_attempt(section: &session::Section) {
986 until("the upload failed", || {
987 section.wake();
988 (section.events().iter()).any(|event| matches!(event, Event::Unreachable(_)))
989 });
990}
991
992#[test]
993fn an_upload_the_provider_dropped_merges_back_once_reads_could_have_caught_up() {
994 let fake = Fake::new(Kind::Stream { truncate: true });
995 notebook(&fake);
996 let documents = fake.documents("root");
997 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
998 let section = open(&notebook, &documents, "Notes.one");
999 fake.set(|state| state.cut = Some((2000, false)));
1000 let id = typed(&section, 0..0, "Cut ");
1001 failed_attempt(&section);
1002 // The document stands as the upload was made on: reads may lag it, so it waits.
1003 for _ in 0..5 {
1004 section.wake();
1005 thread::sleep(Duration::from_millis(50));
1006 }
1007 assert_eq!(kept_images(&fake).len(), 1);
1008 assert_eq!(stored(fake.bytes("Notes.one")), "Original text");
1009 aged(&fake);
1010 published(&section, id);
1011 assert_eq!(stored(fake.bytes("Notes.one")), "Cut Original text");
1012 section.close().unwrap();
1013}
1014
1015#[test]
1016fn a_torn_upload_on_a_provider_writing_in_place_is_put_back() {
1017 for at in [2000, 500] {
1018 let fake = Fake::new(Kind::Stream { truncate: true });
1019 notebook(&fake);
1020 let documents = fake.documents("root");
1021 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
1022 let section = open(&notebook, &documents, "Notes.one");
1023 fake.set(|state| state.cut = Some((at, true)));
1024 let id = typed(&section, 0..0, "Torn ");
1025 published(&section, id);
1026 assert_eq!(stored(fake.bytes("Notes.one")), "Torn Original text");
1027 until("the kept image was retired", || {
1028 section.wake();
1029 unmerged(&fake).is_empty()
1030 });
1031 section.close().unwrap();
1032 }
1033}
1034
1035#[test]
1036fn an_upload_refused_before_it_began_is_retried_not_merged() {
1037 let fake = Fake::new(Kind::Stream { truncate: true });
1038 notebook(&fake);
1039 let documents = fake.documents("root");
1040 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
1041 let section = open(&notebook, &documents, "Notes.one");
1042 fake.set(|state| state.full = true);
1043 let id = typed(&section, 0..0, "Once ");
1044 failed_attempt(&section);
1045 assert!(kept_images(&fake).is_empty());
1046 assert_eq!(section.pending().unwrap().len(), 1);
1047 fake.set(|state| state.full = false);
1048 published(&section, id);
1049 assert_eq!(stored(fake.bytes("Notes.one")), "Once Original text");
1050 section.close().unwrap();
1051}
1052
1053#[test]
1054fn devices_whose_uploads_replace_each_other_converge_with_every_string_once() {
1055 for seed in 1..=4u64 {
1056 let fake = Fake::new(Kind::Seekable);
1057 notebook(&fake);
1058 fake.set(|state| state.delayed = true);
1059 let devices: Vec<_> = (0..3)
1060 .map(|device| {
1061 let documents = fake.device(device).documents("root");
1062 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
1063 open(&notebook, &documents, "Notes.one")
1064 })
1065 .collect();
1066 let mut random = seed;
1067 let mut next = |bound: u64| {
1068 random = random
1069 .wrapping_mul(6364136223846793005)
1070 .wrapping_add(1442695040888963407);
1071 (random >> 33) % bound
1072 };
1073 let mut strings = Vec::new();
1074 for round in 0..6 {
1075 let device = next(3) as usize;
1076 let string = format!("s{seed}r{round}d{device}");
1077 typed(&devices[device], 0..0, &format!("{string} "));
1078 strings.push(string);
1079 devices[device].wake();
1080 thread::sleep(Duration::from_millis(next(150)));
1081 if next(2) == 0 {
1082 fake.publish_staged();
1083 }
1084 }
1085 until("every device converged", || {
1086 for device in &devices {
1087 device.wake();
1088 }
1089 thread::sleep(Duration::from_millis(100));
1090 fake.publish_staged();
1091 let file = stored(fake.bytes("Notes.one"));
1092 devices
1093 .iter()
1094 .all(|device| device.pending().unwrap().is_empty() && text(device) == file)
1095 && (0..3).all(|device| unmerged(&fake.device(device)).is_empty())
1096 });
1097 let arena = onestore::Arena::default();
1098 let mut section = onestore::Section::open(&arena, fake.bytes("Notes.one")).unwrap();
1099 let mut conflicts = String::new();
1100 for (_, pages) in section.conflicts().unwrap() {
1101 for conflict in pages {
1102 conflicts += &paragraph(&section.page(conflict.space).unwrap()).1;
1103 }
1104 }
1105 let file = stored(fake.bytes("Notes.one"));
1106 for string in &strings {
1107 let shown = file.matches(string.as_str()).count();
1108 assert!(
1109 shown <= 1,
1110 "seed {seed}: {string} shows {shown} times in {file}"
1111 );
1112 assert!(
1113 shown == 1 || conflicts.contains(string.as_str()),
1114 "seed {seed}: {string} is gone from {file}"
1115 );
1116 }
1117 for device in devices {
1118 device.close().unwrap();
1119 }
1120 }
1121}
1122
1123#[test]
1124fn a_document_this_device_did_not_write_that_does_not_parse_is_never_rewritten() {
1125 let fake = Fake::new(Kind::Stream { truncate: true });
1126 notebook(&fake);
1127 let documents = fake.documents("root");
1128 let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
1129 let section = open(&notebook, &documents, "Notes.one");
1130 let id = typed(&section, 0..0, "Mine ");
1131 published(&section, id);
1132 // The same document as another writer took it on, cut off, then with checksums that fail.
1133 let mut theirs = fake.bytes("Notes.one");
1134 let arena = onestore::Arena::default();
1135 let mut section_there = onestore::Section::open(&arena, theirs.clone()).unwrap();
1136 let space = section_there.pages().unwrap()[0].0;
1137 let op = PageOp::Text {
1138 text: paragraph(&section_there.page(space).unwrap()).0,
1139 range: 0..0,
1140 with: "Theirs ".into(),
1141 };
1142 let edit = Edit {
1143 at: crate::now(),
1144 ops: vec![Op::Page { space, op }],
1145 };
1146 section_there.apply("Them", &edit).unwrap();
1147 let sealed = section_there.seal().unwrap().unwrap();
1148 sealed.apply(&mut theirs).unwrap();
1149 let mut damaged = theirs.clone();
1150 let at = (1024..damaged.len())
1151 .find(|at| {
1152 let mut flipped = damaged.clone();
1153 flipped[*at] ^= 0xff;
1154 Store::parse(&flipped).is_ok_and(|store| !store.checksum_mismatches.is_empty())
1155 })
1156 .unwrap();
1157 damaged[at] ^= 0xff;
1158 for unreadable in [theirs[..theirs.len() - 100].to_vec(), damaged] {
1159 fs::write(fake.file(&fake.id("Notes.one")), &unreadable).unwrap();
1160 let since = fake.modes().len();
1161 for _ in 0..5 {
1162 section.wake();
1163 thread::sleep(Duration::from_millis(50));
1164 }
1165 status(&section, |status| status.state() == SyncState::InUse);
1166 assert_eq!(fake.bytes("Notes.one"), unreadable);
1167 assert!(fake.modes()[since..].iter().all(|mode| mode == "r"));
1168 }
1169 section.close().unwrap();
1170}