clover's git

snowbound

1	#[path = "../../onestore/tests/support/ops.rs"]
2	mod ops;
3	use notebook::{
4	EditStatus, Recovery, Resolution,
5	session::{Event, Notebook, Section, SyncStatus},
6	};
7	use onestore::{
8	ExGuid,
9	op::{Edit, Op, PageOp},
10	page::Page,
11	};
12	use std::{
13	path::Path,
14	sync::{
15	Arc,
16	atomic::{AtomicUsize, Ordering},
17	},
18	time::{Duration, Instant},
19	};
20
21	#[path = "support/model_ops.rs"]
22	mod model_ops;
23
24	fn first_text(page: &Page) -> ExGuid {
25	page.objects
26	.iter()
27	.find_map(|object| match object {
28	onestore::page::PageObject::Outline(outline) => outline
29	.paragraphs
30	.iter()
31	.find_map(|p| p.text().map(|t| t.id)),
32	_ => None,
33	})
34	.unwrap()
35	}
36
37	fn edited(page: &Page, text: &str) -> Page {
38	let mut after = page.clone();
39	let id = first_text(page);
40	model_ops::replace_text(&mut after, id, 0..0, text);
41	after
42	}
43
44	/// Replaces `range` of the page's first body text as the editor does: the edit's id once
45	/// it is durable.
46	fn typed(
47	section: &Section,
48	space: ExGuid,
49	page: &Page,
50	range: std::ops::Range<u32>,
51	text: &str,
52	) -> u64 {
53	let op = PageOp::Text {
54	text: first_text(page),
55	range,
56	with: text.into(),
57	};
58	let edit = Edit {
59	at: model_ops::now(),
60	ops: vec![Op::Page { space, op }],
61	};
62	section.replica().apply("Editor", edit).unwrap()
63	}
64
65	fn open(file: &Path, cache: &Path) -> (Section, Arc<AtomicUsize>) {
66	let notified = Arc::new(AtomicUsize::new(0));
67	let counter = Arc::clone(&notified);
68	let section = Section::open(file, cache, false, move || {
69	counter.fetch_add(1, Ordering::SeqCst);
70	})
71	.unwrap();
72	(section, notified)
73	}
74
75	fn wait(section: &Section, mut accept: impl FnMut(&Event) -> bool) {
76	let deadline = Instant::now() + Duration::from_secs(20);
77	loop {
78	for event in section.events() {
79	if accept(&event) {
80	return;
81	}
82	}
83	assert!(
84	Instant::now() < deadline,
85	"the expected event did not arrive"
86	);
87	std::thread::sleep(Duration::from_millis(20));
88	}
89	}
90
91	fn published(section: &Section, id: u64) {
92	let deadline = Instant::now() + Duration::from_secs(20);
93	while !matches!(
94	section.status(id).unwrap(),
95	Some(EditStatus::Published { .. })
96	) {
97	assert!(Instant::now() < deadline, "the edit was not published");
98	std::thread::sleep(Duration::from_millis(20));
99	}
100	}
101
102	fn status_until(section: &Section, accept: impl Fn(&SyncStatus) -> bool) -> SyncStatus {
103	let deadline = Instant::now() + Duration::from_secs(20);
104	loop {
105	let status = section.sync_status().unwrap();
106	if accept(&status) {
107	return status;
108	}
109	assert!(Instant::now() < deadline, "the status stayed {status:?}");
110	std::thread::sleep(Duration::from_millis(20));
111	}
112	}
113
114	fn stored_page(file: &Path, space: ExGuid) -> Page {
115	model_ops::page_of(&onestore::read_file(file).unwrap(), space)
116	}
117
118	/// The page title is derived from its content on read; an edited model keeps the old one.
119	fn assert_same(actual: Page, expected: &Page) {
120	let mut expected = expected.clone();
121	expected.title = actual.title.clone();
122	assert_eq!(actual, expected);
123	}
124
125	#[test]
126	fn a_section_is_shared_between_threads() {
127	fn shared<T: Send + Sync>() {}
128	shared::<Section>();
129	shared::<notebook::Replica>();
130	}
131
132	#[test]
133	fn a_section_opens_through_its_replica_and_a_save_reaches_the_file_and_survives_relaunch() {
134	let directory = tempfile::tempdir().unwrap();
135	let file = directory.path().join("notes.one");
136	let cache = directory.path().join("cache");
137	std::fs::write(
138	&file,
139	onestore::create_section("notes.one", "Original", "Author").unwrap(),
140	)
141	.unwrap();
142	let (section, notified) = open(&file, &cache);
143	let pages = section.pages().unwrap();
144	assert_eq!(pages.len(), 1);
145	let space = pages[0].0;
146	let before = section.page(space).unwrap();
147	let after = edited(&before, "Saved ");
148	let id = typed(&section, space, &before, 0..0, "Saved ");
149	assert_same(section.page(space).unwrap(), &after);
150	published(&section, id);
151	// The worker reports the step after writing its receipt.
152	let deadline = Instant::now() + Duration::from_secs(20);
153	while notified.load(Ordering::SeqCst) == 0 {
154	assert!(Instant::now() < deadline, "the host was not woken");
155	std::thread::sleep(Duration::from_millis(10));
156	}
157	assert_same(stored_page(&file, space), &after);
158	assert!(matches!(
159	section.status(id).unwrap(),
160	Some(EditStatus::Published { .. })
161	));
162	section.close().unwrap();
163	let (section, _) = open(&file, &cache);
164	assert_same(section.page(space).unwrap(), &after);
165	assert!(section.pending().unwrap().is_empty());
166	assert!(matches!(
167	section.status(id).unwrap(),
168	Some(EditStatus::Published { .. })
169	));
170	section.close().unwrap();
171	if let Some(destination) = std::env::var_os("ONESTORE_SESSION_NATIVE_EXPORT") {
172	let destination = std::path::PathBuf::from(destination);
173	std::fs::create_dir(&destination).unwrap();
174	let bytes = onestore::read_file(&file).unwrap();
175	let identity = onestore::Store::parse(&bytes).unwrap().header.file_id;
176	std::fs::write(destination.join("notes.one"), bytes).unwrap();
177	std::fs::write(
178	destination.join("Open Notebook.onetoc2"),
179	onestore::create_table_of_contents("Open Notebook.onetoc2", &[("notes.one", identity)])
180	.unwrap(),
181	)
182	.unwrap();
183	}
184	}
185
186	#[test]
187	fn saves_wait_for_an_unreachable_file_and_publish_after_relaunch() {
188	let directory = tempfile::tempdir().unwrap();
189	let file = directory.path().join("notes.one");
190	let cache = directory.path().join("cache");
191	std::fs::write(
192	&file,
193	onestore::create_section("notes.one", "Original", "Author").unwrap(),
194	)
195	.unwrap();
196	let (section, _) = open(&file, &cache);
197	let space = section.pages().unwrap()[0].0;
198	let before = section.page(space).unwrap();
199	let permissions = std::fs::metadata(&file).unwrap().permissions();
200	let mut read_only = permissions.clone();
201	read_only.set_readonly(true);
202	std::fs::set_permissions(&file, read_only).unwrap();
203	let after = edited(&before, "Offline ");
204	let id = typed(&section, space, &before, 0..0, "Offline ");
205	wait(&section, |event| matches!(event, Event::Unreachable(_)));
206	section.close().unwrap();
207	assert_eq!(
208	notebook::Replica::open(
209	walkdir(&cache)
210	.into_iter()
211	.find(|path| path.extension().is_some_and(|ext| ext == "sqlite"))
212	.expect("the section's replica")
213	)
214	.unwrap()
215	.status(id)
216	.unwrap(),
217	Some(EditStatus::Pending)
218	);
219	let (section, _) = open(&file, &cache);
220	assert_same(section.page(space).unwrap(), &after);
221	assert_eq!(section.pending().unwrap().len(), 1);
222	assert_eq!(stored_page(&file, space), before);
223	std::fs::set_permissions(&file, permissions).unwrap();
224	section.wake();
225	published(&section, id);
226	assert_same(stored_page(&file, space), &after);
227	section.close().unwrap();
228	}
229
230	#[test]
231	fn an_external_change_reloads_the_page_and_later_edits_apply_to_it() {
232	let directory = tempfile::tempdir().unwrap();
233	let file = directory.path().join("notes.one");
234	let cache = directory.path().join("cache");
235	std::fs::write(
236	&file,
237	onestore::create_section("notes.one", "Original", "Author").unwrap(),
238	)
239	.unwrap();
240	let (section, _) = open(&file, &cache);
241	let space = section.pages().unwrap()[0].0;
242	let before = section.page(space).unwrap();
243	let native = edited(&before, "Native ");
244	let deadline = Instant::now() + Duration::from_secs(20);
245	loop {
246	let bytes = onestore::read_file(&file).unwrap();
247	match ops::save(&bytes, space, &native)
248	.unwrap()
249	.commit_file(&file)
250	{
251	Ok(()) => break,
252	Err(error) if error.error.kind() == std::io::ErrorKind::WouldBlock => {
253	assert!(Instant::now() < deadline);
254	std::thread::sleep(Duration::from_millis(20));
255	}
256	Err(error) => panic!("{error:?}"),
257	}
258	}
259	section.wake();
260	wait(
261	&section,
262	|event| matches!(event, Event::Changed(spaces) if spaces.contains(&space)),
263	);
264	assert_same(section.page(space).unwrap(), &native);
265	// An edit made on the page shown before the change applies to the page as it is now.
266	let id = typed(&section, space, &before, 0..0, "Local ");
267	let after = edited(&native, "Local ");
268	published(&section, id);
269	assert_same(stored_page(&file, space), &after);
270	section.close().unwrap();
271	}
272
273	#[test]
274	fn edits_apply_in_order_without_waiting_and_finish_before_close() {
275	let directory = tempfile::tempdir().unwrap();
276	let file = directory.path().join("notes.one");
277	let cache = directory.path().join("cache");
278	std::fs::write(
279	&file,
280	onestore::create_section("notes.one", "Original", "Author").unwrap(),
281	)
282	.unwrap();
283	let (section, _) = open(&file, &cache);
284	let space = section.pages().unwrap()[0].0;
285	let original = section.page(space).unwrap();
286	let text = first_text(&original);
287	let apply = |word: &str| {
288	let op = PageOp::Text {
289	text,
290	range: 0..0,
291	with: word.into(),
292	};
293	let edit = Edit {
294	at: model_ops::now(),
295	ops: vec![Op::Page { space, op }],
296	};
297	section.apply("Editor", edit).unwrap();
298	};
299	let mut expected = original.clone();
300	for word in ["one ", "two ", "three ", "four ", "five "] {
301	apply(word);
302	expected = edited(&expected, word);
303	}
304	let deadline = Instant::now() + Duration::from_secs(20);
305	loop {
306	let mut stored = section.page(space).unwrap();
307	stored.title = expected.title.clone();
308	if stored == expected {
309	break;
310	}
311	assert!(Instant::now() < deadline, "the edits did not arrive");
312	std::thread::sleep(Duration::from_millis(20));
313	}
314	while !section.pending().unwrap().is_empty() {
315	assert!(Instant::now() < deadline, "the edits were not published");
316	std::thread::sleep(Duration::from_millis(20));
317	}
318	let native = edited(&section.page(space).unwrap(), "Native ");
319	let deadline = Instant::now() + Duration::from_secs(20);
320	loop {
321	let bytes = onestore::read_file(&file).unwrap();
322	match ops::save(&bytes, space, &native)
323	.unwrap()
324	.commit_file(&file)
325	{
326	Ok(()) => break,
327	Err(error) if error.error.kind() == std::io::ErrorKind::WouldBlock => {
328	assert!(Instant::now() < deadline);
329	std::thread::sleep(Duration::from_millis(20));
330	}
331	Err(error) => panic!("{error:?}"),
332	}
333	}
334	section.wake();
335	wait(
336	&section,
337	|event| matches!(event, Event::Changed(spaces) if spaces.contains(&space)),
338	);
339	apply("Local ");
340	section.close().unwrap();
341	let (section, _) = open(&file, &cache);
342	assert_same(section.page(space).unwrap(), &edited(&native, "Local "));
343	section.close().unwrap();
344	}
345
346	#[test]
347	fn a_notebook_directory_lists_its_sections_and_opens_them() {
348	let directory = tempfile::tempdir().unwrap();
349	let root = directory.path().join("Personal");
350	std::fs::create_dir_all(root.join("Group")).unwrap();
351	for (name, text) in [
352	("Personal/First.one", "One"),
353	("Personal/Group/Second.one", "Two"),
354	] {
355	std::fs::write(
356	directory.path().join(name),
357	onestore::create_section(
358	Path::new(name).file_name().unwrap().to_str().unwrap(),
359	text,
360	"Author",
361	)
362	.unwrap(),
363	)
364	.unwrap();
365	}
366	let notebook = Notebook::open(&root, directory.path().join("cache")).unwrap();
367	let catalog = notebook.catalog();
368	assert_eq!(catalog.sections.len(), 1);
369	assert_eq!(catalog.groups.len(), 1);
370	assert_eq!(catalog.groups[0].sections.len(), 1);
371	let path = catalog.groups[0].sections[0].path.clone();
372	let section = notebook.section(&path, false, || {}).unwrap();
373	let (space, _, _) = section.pages().unwrap()[0];
374	let page = section.page(space).unwrap();
375	let text = first_text(&page);
376	assert_eq!(
377	model_ops::paragraph_with(&page, text)
378	.unwrap()
379	.text()
380	.unwrap()
381	.text
382	.text(),
383	"Two"
384	);
385	section.close().unwrap();
386	}
387
388	/// The replicas in `cache`, in the folder of each location.
389	fn replicas(cache: &Path) -> usize {
390	std::fs::read_dir(cache.join("replicas"))
391	.into_iter()
392	.flatten()
393	.flat_map(|folder| std::fs::read_dir(folder.unwrap().path()).unwrap())
394	.filter(|entry| entry.as_ref().unwrap().path().extension() == Some("sqlite".as_ref()))
395	.count()
396	}
397
398	#[test]
399	fn a_notebook_only_opens_discovered_section_paths() {
400	let directory = tempfile::tempdir().unwrap();
401	let root = directory.path().join("Notebook");
402	let cache = directory.path().join("cache");
403	std::fs::create_dir(&root).unwrap();
404	let source = onestore::create_section("notes.one", "Original", "Author").unwrap();
405	std::fs::write(root.join("notes.one"), &source).unwrap();
406	let outside = directory.path().join("outside.one");
407	std::fs::write(&outside, &source).unwrap();
408	let notebook = Notebook::open(&root, &cache).unwrap();
409	std::fs::write(root.join("added.one"), &source).unwrap();
410	for path in ["../outside.one", outside.to_str().unwrap(), "added.one"] {
411	assert!(matches!(
412	notebook.section(path, false, || {}),
413	Err(notebook::Error::Io(error)) if error.kind() == std::io::ErrorKind::NotFound
414	));
415	}
416	assert_eq!(replicas(&cache), 0);
417	assert_eq!(std::fs::read(&outside).unwrap(), source);
418	}
419
420	#[cfg(unix)]
421	#[test]
422	fn a_catalog_section_replaced_by_an_outside_symlink_is_rejected() {
423	let directory = tempfile::tempdir().unwrap();
424	let root = directory.path().join("Notebook");
425	let cache = directory.path().join("cache");
426	std::fs::create_dir(&root).unwrap();
427	let file = root.join("notes.one");
428	let outside = directory.path().join("outside.one");
429	let source = onestore::create_section("notes.one", "Original", "Author").unwrap();
430	std::fs::write(&file, &source).unwrap();
431	std::fs::write(&outside, &source).unwrap();
432	let notebook = Notebook::open(&root, &cache).unwrap();
433	std::fs::remove_file(&file).unwrap();
434	std::os::unix::fs::symlink(&outside, &file).unwrap();
435	assert!(matches!(
436	notebook.section("notes.one", false, || {}),
437	Err(notebook::Error::Io(error)) if error.kind() == std::io::ErrorKind::PermissionDenied
438	));
439	assert_eq!(replicas(&cache), 0);
440	assert_eq!(std::fs::read(&outside).unwrap(), source);
441	}
442
443	#[test]
444	fn an_absent_remote_does_not_prevent_local_relaunch_or_further_saves() {
445	let directory = tempfile::tempdir().unwrap();
446	let file = directory.path().join("remote.one");
447	let cache = directory.path().join("replica.sqlite");
448	let source = onestore::create_section("remote.one", "Original", "Author").unwrap();
449	let replica = notebook::Replica::create(&cache, &source).unwrap();
450	let section = Section::resume(&file, replica, false, || {}).unwrap();
451	let space = section.pages().unwrap()[0].0;
452	let before = section.page(space).unwrap();
453	let after = edited(&before, "First ");
454	let id = typed(&section, space, &before, 0..0, "First ");
455	wait(
456	&section,
457	|event| matches!(event, Event::Unreachable(error) if error.kind() == std::io::ErrorKind::NotFound),
458	);
459	assert_eq!(section.status(id).unwrap(), Some(EditStatus::Pending));
460	section.close().unwrap();
461	assert!(!file.exists());
462
463	let section = Section::resume(
464	&file,
465	notebook::Replica::open(&cache).unwrap(),
466	false,
467	|| {},
468	)
469	.unwrap();
470	assert_same(section.page(space).unwrap(), &after);
471	assert_eq!(section.status(id).unwrap(), Some(EditStatus::Pending));
472	let before = section.page(space).unwrap();
473	let after = edited(&before, "Second ");
474	let next = typed(&section, space, &before, 0..0, "Second ");
475	assert!(!file.exists());
476	let restored = directory.path().join("restored.one");
477	std::fs::write(&restored, &source).unwrap();
478	std::fs::rename(restored, &file).unwrap();
479	section.wake();
480	published(&section, next);
481	assert_same(stored_page(&file, space), &after);
482	section.close().unwrap();
483	let section = Section::resume(
484	&file,
485	notebook::Replica::open(&cache).unwrap(),
486	false,
487	|| {},
488	)
489	.unwrap();
490	assert_same(section.page(space).unwrap(), &after);
491	assert!(section.pending().unwrap().is_empty());
492	section.close().unwrap();
493	}
494
495	#[test]
496	fn resuming_against_another_document_preserves_both_remote_and_pending_edit() {
497	let directory = tempfile::tempdir().unwrap();
498	let file = directory.path().join("other.one");
499	let cache = directory.path().join("replica.sqlite");
500	let source = onestore::create_section("notes.one", "Original", "Author").unwrap();
501	let replica = notebook::Replica::create(&cache, &source).unwrap();
502	let section = Section::resume(&file, replica, false, || {}).unwrap();
503	let space = section.pages().unwrap()[0].0;
504	let before = section.page(space).unwrap();
505	let after = edited(&before, "Local ");
506	let id = typed(&section, space, &before, 0..0, "Local ");
507	wait(&section, |event| matches!(event, Event::Unreachable(_)));
508	section.close().unwrap();
509
510	let other = onestore::create_section("other.one", "Unrelated", "Other").unwrap();
511	std::fs::write(&file, &other).unwrap();
512	let section = Section::resume(
513	&file,
514	notebook::Replica::open(&cache).unwrap(),
515	false,
516	|| {},
517	)
518	.unwrap();
519	wait(&section, |event| matches!(event, Event::Failed(_)));
520	assert_eq!(section.status(id).unwrap(), Some(EditStatus::Pending));
521	assert_same(section.page(space).unwrap(), &after);
522	assert_eq!(std::fs::read(&file).unwrap(), other);
523	assert!(section.close().is_err());
524	let replica = notebook::Replica::open(&cache).unwrap();
525	assert_eq!(replica.status(id).unwrap(), Some(EditStatus::Pending));
526	assert_eq!(std::fs::read(&file).unwrap(), other);
527	}
528
529	#[test]
530	fn offline_save_process() {
531	let Some(root) = std::env::var_os("ONENOTE_SESSION_CHILD") else {
532	return;
533	};
534	let root = std::path::PathBuf::from(root);
535	let replica = notebook::Replica::open(root.join("replica.sqlite")).unwrap();
536	let section = Section::resume(root.join("absent.one"), replica, false, || {}).unwrap();
537	let space = section.pages().unwrap()[0].0;
538	let before = section.page(space).unwrap();
539	let id = typed(&section, space, &before, 0..0, "Durable ");
540	std::fs::write(root.join("acknowledged"), id.to_string()).unwrap();
541	// Terminate without running Session or SQLite destructors after acknowledgement.
542	std::process::exit(0);
543	}
544
545	#[test]
546	fn an_acknowledged_offline_save_survives_process_exit_without_cleanup() {
547	let directory = tempfile::tempdir().unwrap();
548	let cache = directory.path().join("replica.sqlite");
549	let source = onestore::create_section("absent.one", "Original", "Author").unwrap();
550	drop(notebook::Replica::create(&cache, &source).unwrap());
551	let output = std::process::Command::new(std::env::current_exe().unwrap())
552	.args(["--exact", "offline_save_process", "--nocapture"])
553	.env("ONENOTE_SESSION_CHILD", directory.path())
554	.output()
555	.unwrap();
556	assert!(output.status.success(), "{output:?}");
557	let id: u64 = std::fs::read_to_string(directory.path().join("acknowledged"))
558	.unwrap()
559	.parse()
560	.unwrap();
561	let file = directory.path().join("absent.one");
562	assert!(!file.exists());
563	let section = Section::resume(
564	&file,
565	notebook::Replica::open(&cache).unwrap(),
566	false,
567	|| {},
568	)
569	.unwrap();
570	let space = section.pages().unwrap()[0].0;
571	let original = model_ops::page_of(&source, space);
572	let after = edited(&original, "Durable ");
573	assert_same(section.page(space).unwrap(), &after);
574	assert_eq!(section.status(id).unwrap(), Some(EditStatus::Pending));
575	section.close().unwrap();
576	}
577
578	#[cfg(feature = "smb")]
579	#[test]
580	fn smb_connection_failure_keeps_the_session_locally_editable_and_retries() {
581	let directory = tempfile::tempdir().unwrap();
582	let cache = directory.path().join("replica.sqlite");
583	let source = onestore::create_section("notes.one", "Original", "Author").unwrap();
584	let replica = notebook::Replica::create(&cache, &source).unwrap();
585	let attempts = Arc::new(AtomicUsize::new(0));
586	let counter = Arc::clone(&attempts);
587	let section = Section::resume_smb(
588	"Folder/notes.one".into(),
589	replica,
590	1024 * 1024,
591	move || {
592	counter.fetch_add(1, Ordering::SeqCst);
593	Err(std::io::ErrorKind::PermissionDenied.into())
594	},
595	false,
596	|| {},
597	)
598	.unwrap();
599	wait(
600	&section,
601	|event| matches!(event, Event::Unreachable(error) if error.kind() == std::io::ErrorKind::PermissionDenied),
602	);
603	let space = section.pages().unwrap()[0].0;
604	let before = section.page(space).unwrap();
605	let after = edited(&before, "Offline SMB ");
606	let id = typed(&section, space, &before, 0..0, "Offline SMB ");
607	section.wake();
608	wait(
609	&section,
610	|event| matches!(event, Event::Unreachable(error) if error.kind() == std::io::ErrorKind::PermissionDenied),
611	);
612	assert!(attempts.load(Ordering::SeqCst) >= 2);
613	assert_eq!(section.status(id).unwrap(), Some(EditStatus::Pending));
614	assert_same(section.page(space).unwrap(), &after);
615	assert_eq!(section.file(), Path::new("Folder/notes.one"));
616	section.close().unwrap();
617	assert_eq!(
618	notebook::Replica::open(&cache).unwrap().status(id).unwrap(),
619	Some(EditStatus::Pending)
620	);
621	}
622
623	#[cfg(feature = "smb")]
624	#[test]
625	#[ignore = "requires ONESTORE_SESSION_SMB address and a disposable session.one on its agent share"]
626	fn live_smb_session_save_publishes_and_reopens() {
627	use notebook::smb::{Client, Credentials};
628	let address = std::env::var("ONESTORE_SESSION_SMB").unwrap();
629	let connect = move || {
630	Client::connect(
631	&address,
632	"agent",
633	Credentials::default(),
634	Duration::from_secs(5),
635	)
636	};
637	let client = connect().unwrap();
638	let source = client.read("session.one", 1024 * 1024).unwrap();
639	let directory = tempfile::tempdir().unwrap();
640	let cache = directory.path().join("replica.sqlite");
641	let replica = notebook::Replica::create(&cache, &source).unwrap();
642	let section = Section::resume_smb(
643	"session.one".into(),
644	replica,
645	1024 * 1024,
646	connect.clone(),
647	false,
648	|| {},
649	)
650	.unwrap();
651	let space = section.pages().unwrap()[0].0;
652	let before = section.page(space).unwrap();
653	let after = edited(&before, "Session SMB ");
654	let id = typed(&section, space, &before, 0..0, "Session SMB ");
655	published(&section, id);
656	let remote = client.read("session.one", 1024 * 1024).unwrap();
657	assert_same(model_ops::page_of(&remote, space), &after);
658	section.close().unwrap();
659	let section = Section::resume_smb(
660	"session.one".into(),
661	notebook::Replica::open(&cache).unwrap(),
662	1024 * 1024,
663	connect,
664	false,
665	|| {},
666	)
667	.unwrap();
668	assert_same(section.page(space).unwrap(), &after);
669	assert!(section.pending().unwrap().is_empty());
670	assert!(matches!(
671	section.status(id).unwrap(),
672	Some(EditStatus::Published { .. })
673	));
674	section.close().unwrap();
675	}
676
677	#[cfg(feature = "smb")]
678	#[test]
679	fn dropping_during_connection_keeps_cache_owned_until_the_worker_finishes() {
680	use std::sync::mpsc;
681	let directory = tempfile::tempdir().unwrap();
682	let cache = directory.path().join("replica.sqlite");
683	let source = onestore::create_section("notes.one", "Original", "Author").unwrap();
684	let replica = notebook::Replica::create(&cache, &source).unwrap();
685	let (entered, connecting) = mpsc::channel();
686	let (release, stalled) = mpsc::channel();
687	let section = Section::resume_smb(
688	"notes.one".into(),
689	replica,
690	1024 * 1024,
691	move || {
692	entered.send(()).unwrap();
693	stalled.recv_timeout(Duration::from_secs(120)).unwrap();
694	Err(std::io::ErrorKind::TimedOut.into())
695	},
696	false,
697	|| {},
698	)
699	.unwrap();
700	connecting.recv_timeout(Duration::from_secs(120)).unwrap();
701	let space = section.pages().unwrap()[0].0;
702	let before = section.page(space).unwrap();
703	let after = edited(&before, "Saved during connection ");
704	let id = typed(&section, space, &before, 0..0, "Saved during connection ");
705	let (dropped, finished) = mpsc::channel();
706	let owner = std::thread::spawn(move || {
707	drop(section);
708	dropped.send(()).unwrap();
709	});
710	finished.recv_timeout(Duration::from_secs(60)).unwrap();
711	assert!(matches!(
712	notebook::Replica::open(&cache),
713	Err(notebook::Error::Database(error))
714	if error.sqlite_error_code() == Some(rusqlite::ErrorCode::DatabaseBusy)
715	));
716	release.send(()).unwrap();
717	owner.join().unwrap();
718	let deadline = Instant::now() + Duration::from_secs(5);
719	let replica = loop {
720	match notebook::Replica::open(&cache) {
721	Ok(replica) => break replica,
722	Err(notebook::Error::Database(error))
723	if error.sqlite_error_code() == Some(rusqlite::ErrorCode::DatabaseBusy) =>
724	{
725	assert!(Instant::now() < deadline, "worker retained cache ownership");
726	std::thread::sleep(Duration::from_millis(10));
727	}
728	Err(error) => panic!("{error}"),
729	}
730	};
731	assert_eq!(replica.status(id).unwrap(), Some(EditStatus::Pending));
732	assert_same(
733	model_ops::page_of(&server::snapshot(&replica), space),
734	&after,
735	);
736	assert!(connecting.try_recv().is_err());
737	}
738
739	#[test]
740	fn a_conflicting_save_keeps_the_native_page_and_a_conflict_page_the_session_deletes() {
741	let directory = tempfile::tempdir().unwrap();
742	let file = directory.path().join("notes.one");
743	let cache = directory.path().join("cache");
744	std::fs::write(
745	&file,
746	onestore::create_section("notes.one", "Original", "Author").unwrap(),
747	)
748	.unwrap();
749	let (section, _) = open(&file, &cache);
750	let space = section.pages().unwrap()[0].0;
751	let before = section.page(space).unwrap();
752	let text = first_text(&before);
753	let permissions = std::fs::metadata(&file).unwrap().permissions();
754	let mut read_only = permissions.clone();
755	read_only.set_readonly(true);
756	std::fs::set_permissions(&file, read_only).unwrap();
757	let mut local = before.clone();
758	model_ops::replace_text(&mut local, text, 0..8, "Local");
759	let id = typed(&section, space, &before, 0..8, "Local");
760	wait(&section, |event| matches!(event, Event::Unreachable(_)));
761	std::fs::set_permissions(&file, permissions).unwrap();
762	let mut native = before.clone();
763	model_ops::replace_text(&mut native, text, 0..8, "Native");
764	let deadline = Instant::now() + Duration::from_secs(20);
765	loop {
766	let bytes = onestore::read_file(&file).unwrap();
767	match ops::save(&bytes, space, &native)
768	.unwrap()
769	.commit_file(&file)
770	{
771	Ok(()) => break,
772	Err(error) if error.error.kind() == std::io::ErrorKind::WouldBlock => {
773	assert!(Instant::now() < deadline);
774	std::thread::sleep(Duration::from_millis(20));
775	}
776	Err(error) => panic!("{error:?}"),
777	}
778	}
779	section.wake();
780	wait(
781	&section,
782	|event| matches!(event, Event::Attempt { id: n, status: EditStatus::Published { .. } } if *n > id),
783	);
784	assert!(matches!(
785	section.status(id).unwrap(),
786	Some(EditStatus::Published { .. })
787	));
788	// As OneNote does, the native version stays the page and the local one is kept beside it.
789	assert_same(section.page(space).unwrap(), &native);
790	assert_same(stored_page(&file, space), &native);
791	let listed = section.conflicts().unwrap();
792	assert_eq!(listed.len(), 1);
793	assert_eq!(listed[0].0, space);
794	let conflict = &listed[0].1[0];
795	assert_eq!(conflict.user, "Editor");
796	let kept = section.page(conflict.space).unwrap();
797	assert_eq!(
798	model_ops::paragraph_with(&kept, first_text(&kept))
799	.and_then(|paragraph| paragraph.text())
800	.map(|text| text.text.text().to_owned()),
801	Some("Local".to_owned())
802	);
803	assert_eq!(
804	server::conflicts(&onestore::read_file(&file).unwrap())[0].1,
805	[("Editor".to_owned(), vec!["Local".to_owned()])]
806	);
807	// Merged by hand, the conflict page is deleted.
808	let deleted = section.delete_pages(&[conflict.space]).unwrap();
809	published(&section, deleted);
810	assert!(section.conflicts().unwrap().is_empty());
811	assert!(server::conflicts(&onestore::read_file(&file).unwrap()).is_empty());
812	assert!(section.pending().unwrap().is_empty());
813	section.close().unwrap();
814	}
815
816	#[path = "support/server.rs"]
817	mod server;
818
819	/// An uncertain attempt survives a restart as `AwaitingConfirmation`; after exporting the
820	/// archive the user either publishes the local edits again or abandons the branch. Neither
821	/// path records a receipt for the uncertain attempt.
822	#[test]
823	fn an_uncertain_attempt_is_released_after_restart_by_review() {
824	for continued in [true, false] {
825	let directory = tempfile::tempdir().unwrap();
826	let file = directory.path().join("notes.one");
827	let cache = directory.path().join("cache.sqlite");
828	let source = onestore::create_section("notes.one", "Original", "Author").unwrap();
829	std::fs::write(&file, &source).unwrap();
830	let replica = notebook::Replica::create(&cache, &source).unwrap();
831	let space = space_of(&source);
832	let page = model_ops::page_of(&source, space);
833	let local = edited(&page, "Uncertain ");
834	let id = model_ops::save_as(&replica, space, &local, "Author")
835	.unwrap()
836	.unwrap();
837	let mut faulty = server::Server::new(&source);
838	faulty.fault = server::Fault::UnknownBefore;
839	assert!(replica.sync_once(&mut faulty).is_err());
840	let attempted = replica.status(id).unwrap();
841	assert!(matches!(
842	attempted,
843	Some(EditStatus::AwaitingConfirmation { .. })
844	));
845	drop(replica);
846
847	let section =
848	Section::resume(&file, notebook::Replica::open(&cache).unwrap(), true, || {}).unwrap();
849	assert_eq!(section.status(id).unwrap(), attempted);
850	assert_eq!(
851	section
852	.pending()
853	.unwrap()
854	.iter()
855	.map(|edit| edit.id)
856	.collect::<Vec<_>>(),
857	[id]
858	);
859	assert_same(section.page(space).unwrap(), &local);
860	let archive = directory.path().join("review.sqlite");
861	if continued {
862	section.release(id, &archive, Resolution::Mine).unwrap();
863	assert_eq!(section.status(id).unwrap(), Some(EditStatus::Pending));
864	section.set_offline(false);
865	published(&section, id);
866	assert_same(stored_page(&file, space), &local);
867	} else {
868	section.release(id, &archive, Resolution::Theirs).unwrap();
869	assert_eq!(
870	section.status(id).unwrap(),
871	Some(EditStatus::Archived {
872	archive: archive.to_string_lossy().into_owned()
873	})
874	);
875	assert_same(section.page(space).unwrap(), &page);
876	assert_eq!(stored_page(&file, space), page);
877	}
878	let recovery = Recovery::open(&archive).unwrap();
879	assert_eq!(recovery.pending().unwrap().len(), 1);
880	assert_eq!(recovery.status(id).unwrap(), attempted);
881	assert!(section.pending().unwrap().is_empty());
882	assert!(
883	section
884	.release(
885	id,
886	directory.path().join("again.sqlite"),
887	Resolution::Theirs
888	)
889	.is_err()
890	);
891	section.close().unwrap();
892	}
893	}
894
895	fn space_of(source: &[u8]) -> ExGuid {
896	let store = onestore::Store::parse(source).unwrap();
897	let index = onestore::RevisionIndex::parse(&store).unwrap();
898	onestore::document::Document::parse(&index)
899	.unwrap()
900	.pages()
901	.unwrap()[0]
902	.0
903	}
904
905	/// Restoring and deleting page versions go through the queue and reach the file as OneNote
906	/// 2010 stores them (`corpus/page-versions`); a publication that changes only a page's
907	/// versions confirms by the history revision it wrote.
908	#[test]
909	fn page_versions_restore_and_delete_through_the_session() {
910	let directory = tempfile::tempdir().unwrap();
911	let file = directory.path().join("History.one");
912	let cache = directory.path().join("cache");
913	std::fs::copy(
914	Path::new(env!("CARGO_MANIFEST_DIR"))
915	.join("../../corpus/page-versions/native/step-02/notebook/History.one"),
916	&file,
917	)
918	.unwrap();
919	let (section, _) = open(&file, &cache);
920	let listed = section.versions().unwrap();
921	let (space, versions) = listed[0].clone();
922	assert_eq!(versions.len(), 1);
923	let standing = section.page(space).unwrap();
924	let old = section.version(space, versions[0].context).unwrap();
925	let restored = section
926	.restore_version(space, versions[0].context, "Editor")
927	.unwrap();
928	published(&section, restored);
929	assert_eq!(section.page(space).unwrap().objects, old.objects);
930	let stored = |file: &Path| {
931	let arena = onestore::Arena::default();
932	let mut stored =
933	onestore::Section::open(&arena, onestore::read_file(file).unwrap()).unwrap();
934	let versions = stored.versions().unwrap();
935	(stored.page(space).unwrap(), versions)
936	};
937	let (page, after) = stored(&file);
938	assert_eq!(page.objects, old.objects);
939	assert_eq!(after[0].1.len(), 2);
940	assert_eq!(after[0].1[0].author.as_deref(), Some("Other Person"));
941	assert_eq!(
942	section.version(space, after[0].1[0].context).unwrap(),
943	standing
944	);
945	section.close().unwrap();
946
947	// After a relaunch, deleting every version publishes the history alone.
948	let (section, _) = open(&file, &cache);
949	let all: Vec<ExGuid> = section.versions().unwrap()[0]
950	.1
951	.iter()
952	.map(|version| version.context)
953	.collect();
954	let deleted = section.delete_versions(&[(space, all)]).unwrap();
955	published(&section, deleted);
956	assert!(section.versions().unwrap().is_empty());
957	let (page, after) = stored(&file);
958	assert!(after.is_empty());
959	assert_eq!(page.objects, old.objects);
960	section.close().unwrap();
961	}
962
963	#[test]
964	fn working_offline_queues_edits_until_sync_now_or_working_online() {
965	let directory = tempfile::tempdir().unwrap();
966	let file = directory.path().join("notes.one");
967	let cache = directory.path().join("cache");
968	std::fs::write(
969	&file,
970	onestore::create_section("notes.one", "Original", "Author").unwrap(),
971	)
972	.unwrap();
973	let (section, _) = open(&file, &cache);
974	let space = section.pages().unwrap()[0].0;
975	let before = section.page(space).unwrap();
976	let reached = status_until(&section, |status| status.synced.is_some());
977	assert!(reached.error.is_none());
978	assert_eq!(reached.queued, 0);
979
980	section.set_offline(true);
981	let id = typed(&section, space, &before, 0..0, "Offline ");
982	std::thread::sleep(Duration::from_millis(500));
983	assert_eq!(section.status(id).unwrap(), Some(EditStatus::Pending));
984	assert_eq!(section.sync_status().unwrap().queued, 1);
985	assert_eq!(stored_page(&file, space), before);
986
987	// Sync Now publishes while working offline.
988	section.wake();
989	published(&section, id);
990	let after = edited(&before, "Offline ");
991	assert_same(stored_page(&file, space), &after);
992	let synced = status_until(&section, |status| status.queued == 0);
993	assert!(synced.synced > reached.synced);
994
995	let next = typed(&section, space, &after, 0..0, "Again ");
996	std::thread::sleep(Duration::from_millis(500));
997	assert_eq!(section.status(next).unwrap(), Some(EditStatus::Pending));
998	section.set_offline(false);
999	published(&section, next);
1000	status_until(&section, |status| status.queued == 0);
1001	section.close().unwrap();
1002	}
1003
1004	#[test]
1005	fn a_missing_section_file_reports_its_error_until_it_returns() {
1006	let directory = tempfile::tempdir().unwrap();
1007	let file = directory.path().join("notes.one");
1008	let away = directory.path().join("away.one");
1009	let cache = directory.path().join("cache");
1010	std::fs::write(
1011	&file,
1012	onestore::create_section("notes.one", "Original", "Author").unwrap(),
1013	)
1014	.unwrap();
1015	let (section, notified) = open(&file, &cache);
1016	status_until(&section, |status| status.synced.is_some());
1017	std::fs::rename(&file, &away).unwrap();
1018	section.wake();
1019	let failed = status_until(&section, |status| status.error.is_some());
1020	assert_eq!(
1021	failed.error.map(|error| error.kind()),
1022	Some(std::io::ErrorKind::NotFound)
1023	);
1024	let before = notified.load(Ordering::SeqCst);
1025	std::fs::rename(&away, &file).unwrap();
1026	section.wake();
1027	let back = status_until(&section, |status| status.error.is_none());
1028	assert!(back.synced > failed.synced);
1029	assert!(notified.load(Ordering::SeqCst) > before);
1030	section.close().unwrap();
1031	}
1032
1033	/// A notebook folder holding `First.one` and `Second.one`, and a cache beside it.
1034	fn two_sections(directory: &Path) -> Notebook {
1035	let root = directory.join("Shared");
1036	std::fs::create_dir(&root).unwrap();
1037	for name in ["First", "Second"] {
1038	let file = format!("{name}.one");
1039	std::fs::write(
1040	root.join(&file),
1041	onestore::create_section(&file, name, "Author").unwrap(),
1042	)
1043	.unwrap();
1044	}
1045	Notebook::open(&root, directory.join("cache")).unwrap()
1046	}
1047
1048	fn until(what: &str, mut accept: impl FnMut() -> bool) {
1049	let deadline = Instant::now() + Duration::from_secs(20);
1050	while !accept() {
1051	assert!(Instant::now() < deadline, "{what}");
1052	std::thread::sleep(Duration::from_millis(20));
1053	}
1054	}
1055
1056	#[test]
1057	fn a_closed_sections_queued_edits_publish_in_the_background_when_online() {
1058	let directory = tempfile::tempdir().unwrap();
1059	let mut notebook = two_sections(directory.path());
1060	let file = directory.path().join("Shared/Second.one");
1061	let background = notebook.background(false, false, || {}).unwrap();
1062	background.set_offline(true);
1063	background.watch(notebook.replicas());
1064	// The round started with the background ends before the edit exists.
1065	std::thread::sleep(Duration::from_millis(200));
1066	let section = notebook.section("Second.one", false, || {}).unwrap();
1067	section.set_offline(true);
1068	let space = section.pages().unwrap()[0].0;
1069	let before = section.page(space).unwrap();
1070	typed(&section, space, &before, 0..0, "Closed ");
1071	section.close().unwrap();
1072	std::thread::sleep(Duration::from_millis(300));
1073	assert_eq!(
1074	stored_page(&file, space),
1075	before,
1076	"working offline publishes nothing"
1077	);
1078	background.set_offline(false);
1079	let mut after = edited(&before, "Closed ");
1080	until("the closed section published", || {
1081	let stored = stored_page(&file, space);
1082	after.title = stored.title.clone();
1083	stored == after
1084	});
1085	until("the status shows nothing waiting", || {
1086	background.status().iter().any(|(path, status)| {
1087	path == "Second.one" && status.synced.is_some() && status.queued == 0
1088	})
1089	});
1090	drop(background);
1091	let mut reopened = None;
1092	until("the background released the replica", || {
1093	reopened = notebook.section("Second.one", false, || {}).ok();
1094	reopened.is_some()
1095	});
1096	let reopened = reopened.unwrap();
1097	assert!(reopened.pending().unwrap().is_empty());
1098	reopened.close().unwrap();
1099	}
1100
1101	#[test]
1102	fn a_remote_change_to_a_closed_section_is_noticed_and_rebases_its_replica() {
1103	let directory = tempfile::tempdir().unwrap();
1104	let mut notebook = two_sections(directory.path());
1105	// Second has a replica from being opened once; First has never been opened.
1106	notebook
1107	.section("Second.one", false, || {})
1108	.unwrap()
1109	.close()
1110	.unwrap();
1111	let notified = Arc::new(AtomicUsize::new(0));
1112	let counter = Arc::clone(&notified);
1113	let background = notebook
1114	.background(true, false, move || {
1115	counter.fetch_add(1, Ordering::SeqCst);
1116	})
1117	.unwrap();
1118	background.watch(notebook.replicas());
1119	until("both sections were reached", || {
1120	background
1121	.status()
1122	.iter()
1123	.all(|(_, status)| status.synced.is_some() && status.error.is_none())
1124	});
1125	assert!(background.changed().is_empty());
1126	let mut changed = Vec::new();
1127	for name in ["First.one", "Second.one"] {
1128	let file = directory.path().join("Shared").join(name);
1129	let bytes = onestore::read_file(&file).unwrap();
1130	let space = notebook::session::stored_pages(&bytes).unwrap()[0].space;
1131	let native = edited(&model_ops::page_of(&bytes, space), "Native ");
1132	ops::save(&bytes, space, &native)
1133	.unwrap()
1134	.commit_file(&file)
1135	.unwrap();
1136	changed.push((name, space, native));
1137	}
1138	// As a watch reports a folder when it cannot name the file.
1139	background.touched(&[String::new()]);
1140	let mut noticed = Vec::new();
1141	until("both changes were noticed", || {
1142	noticed.extend(background.changed());
1143	changed
1144	.iter()
1145	.all(|(name, ..)| noticed.iter().any(|path| path == name))
1146	});
1147	assert!(notified.load(Ordering::SeqCst) > 0);
1148	drop(background);
1149	let (_, space, native) = &changed[1];
1150	let replica = notebook.replica_path("Second.one").unwrap();
1151	let mut opened = None;
1152	until("the background released the replica", || {
1153	opened = notebook::Replica::open(&replica).ok();
1154	opened.is_some()
1155	});
1156	assert_same(opened.unwrap().page(*space).unwrap(), native);
1157	}
1158
1159	#[test]
1160	fn a_reported_change_wakes_its_section_without_polling() {
1161	let directory = tempfile::tempdir().unwrap();
1162	let mut notebook = two_sections(directory.path());
1163	let background = notebook.background(true, false, || {}).unwrap();
1164	background.watch(notebook.replicas());
1165	until("both sections were reached", || {
1166	background
1167	.status()
1168	.iter()
1169	.all(|(_, status)| status.synced.is_some())
1170	});
1171	let file = directory.path().join("Shared/Second.one");
1172	let bytes = onestore::read_file(&file).unwrap();
1173	let space = notebook::session::stored_pages(&bytes).unwrap()[0].space;
1174	let native = edited(&model_ops::page_of(&bytes, space), "Native ");
1175	ops::save(&bytes, space, &native)
1176	.unwrap()
1177	.commit_file(&file)
1178	.unwrap();
1179	std::thread::sleep(Duration::from_secs(2));
1180	assert!(
1181	background.changed().is_empty(),
1182	"nothing polls an idle notebook"
1183	);
1184	background.touched(&["Second.one".to_owned()]);
1185	let mut noticed = Vec::new();
1186	until("the reported change was noticed", || {
1187	noticed.extend(background.changed());
1188	!noticed.is_empty()
1189	});
1190	assert_eq!(noticed, ["Second.one"]);
1191	}
1192
1193	#[test]
1194	fn every_section_gets_an_offline_copy_that_closing_the_notebook_discards() {
1195	let directory = tempfile::tempdir().unwrap();
1196	let mut notebook = two_sections(directory.path());
1197	let replicas: Vec<_> = notebook
1198	.replicas()
1199	.into_iter()
1200	.map(|known| known.replica.unwrap())
1201	.collect();
1202	assert!(replicas.iter().all(|replica| !replica.exists()));
1203	let background = notebook.background(true, true, || {}).unwrap();
1204	background.watch(notebook.replicas());
1205	until("both sections have offline copies", || {
1206	replicas.iter().all(|replica| replica.exists())
1207	&& background
1208	.status()
1209	.iter()
1210	.all(|(_, status)| status.synced.is_some())
1211	});
1212	// Opening a section starts from its copy, and edits made offline wait in it.
1213	let section = notebook.section("Second.one", false, || {}).unwrap();
1214	section.set_offline(true);
1215	let space = section.pages().unwrap()[0].0;
1216	let before = section.page(space).unwrap();
1217	typed(&section, space, &before, 0..0, "Waiting ");
1218	section.close().unwrap();
1219	background.discard();
1220	let second = notebook.replica_path("Second.one").unwrap();
1221	until("the copy with nothing waiting was discarded", || {
1222	!notebook.replica_path("First.one").unwrap().exists()
1223	});
1224	assert!(second.exists(), "a copy with edits waiting stays");
1225	// The stopping thread may still be peeking at it, as a host opening a section waits out.
1226	let mut kept = None;
1227	until("the background let go of the copy", || {
1228	match notebook::Replica::open(&second) {
1229	Ok(replica) => kept = Some(replica),
1230	Err(error) if error.busy() => {}
1231	Err(error) => panic!("{error}"),
1232	}
1233	kept.is_some()
1234	});
1235	assert_eq!(kept.unwrap().pending().unwrap().len(), 1);
1236	}
1237
1238	/// A section file whose stamp reads are counted.
1239	struct Counted(std::path::PathBuf, Arc<AtomicUsize>);
1240
1241	#[test]
1242	fn a_moved_section_keeps_its_replica_while_the_old_remote_is_stalled() {
1243	use std::sync::{Condvar, Mutex, mpsc};
1244
1245	struct Stalled {
1246	file: std::path::PathBuf,
1247	source: Vec<u8>,
1248	entered: Arc<Mutex<Option<mpsc::Sender<()>>>>,
1249	release: Arc<(Mutex<bool>, Condvar)>,
1250	finished: Option<mpsc::Sender<()>>,
1251	publications: Arc<AtomicUsize>,
1252	late_error: bool,
1253	}
1254
1255	impl Drop for Stalled {
1256	fn drop(&mut self) {
1257	if let Some(finished) = self.finished.take() {
1258	let _ = finished.send(());
1259	}
1260	}
1261	}
1262
1263	impl notebook::Remote for Stalled {
1264	fn read(&mut self) -> std::io::Result<Vec<u8>> {
1265	Ok(self.source.clone())
1266	}
1267
1268	fn stamp(&mut self) -> std::io::Result<onestore::Stamp> {
1269	if let Some(entered) = self.entered.lock().unwrap().take() {
1270	entered.send(()).unwrap();
1271	let (lock, changed) = &*self.release;
1272	let _ = changed
1273	.wait_timeout_while(lock.lock().unwrap(), Duration::from_secs(5), |done| !*done)
1274	.unwrap();
1275	if self.late_error {
1276	return Err(std::io::ErrorKind::TimedOut.into());
1277	}
1278	}
1279	onestore::Stamp::of(&self.source).map_err(std::io::Error::other)
1280	}
1281
1282	fn publish(
1283	&mut self,
1284	transaction: &onestore::Transaction,
1285	) -> Result<(), onestore::CommitError> {
1286	self.publications.fetch_add(1, Ordering::SeqCst);
1287	transaction.commit_file(&self.file)
1288	}
1289
1290	fn confirm(&mut self, _: &onestore::Stamp) -> Result<(), onestore::CommitError> {
1291	Ok(())
1292	}
1293	}
1294
1295	for late_error in [false, true] {
1296	let directory = tempfile::tempdir().unwrap();
1297	let file = directory.path().join("Before.one");
1298	let target = directory.path().join("After.one");
1299	let source = onestore::create_section("Before.one", "Original", "Author").unwrap();
1300	std::fs::write(&file, &source).unwrap();
1301	let (entered, waiting) = mpsc::channel();
1302	let (connected, connecting) = mpsc::channel();
1303	let (finished, stopped) = mpsc::channel();
1304	let begin = Arc::new((Mutex::new(false), Condvar::new()));
1305	let starting = Arc::clone(&begin);
1306	let release = Arc::new((Mutex::new(false), Condvar::new()));
1307	let gate = Arc::clone(&release);
1308	let entered = Arc::new(Mutex::new(Some(entered)));
1309	let publications = Arc::new(AtomicUsize::new(0));
1310	let attempts = Arc::clone(&publications);
1311	let mut opening = true;
1312	let mut section = Section::open_with(
1313	&file,
1314	directory.path().join("cache"),
1315	move |file| {
1316	let finishing = (!std::mem::replace(&mut opening, false)).then(|| finished.clone());
1317	if finishing.is_some() {
1318	connected.send(()).unwrap();
1319	let (lock, changed) = &*starting;
1320	let _ = changed
1321	.wait_timeout_while(lock.lock().unwrap(), Duration::from_secs(5), |done| {
1322	!*done
1323	})
1324	.unwrap();
1325	}
1326	Ok(Stalled {
1327	file: file.to_owned(),
1328	source: source.clone(),
1329	entered: Arc::clone(&entered),
1330	release: Arc::clone(&gate),
1331	finished: finishing,
1332	publications: Arc::clone(&attempts),
1333	late_error,
1334	})
1335	},
1336	false,
1337	|| {},
1338	)
1339	.unwrap();
1340	connecting.recv_timeout(Duration::from_secs(3)).unwrap();
1341	let replica = Arc::clone(section.replica());
1342	let (space, ..) = section.pages().unwrap()[0];
1343	let before = section.page(space).unwrap();
1344	let edit = typed(&section, space, &before, 0..0, "Moved ");
1345	*begin.0.lock().unwrap() = true;
1346	begin.1.notify_all();
1347	waiting.recv_timeout(Duration::from_secs(3)).unwrap();
1348	section.set_offline(true);
1349	std::fs::rename(&file, &target).unwrap();
1350	let start = Instant::now();
1351	section.relocate(&target).unwrap();
1352	assert!(start.elapsed() < Duration::from_secs(1));
1353	assert!(Arc::ptr_eq(&replica, section.replica()));
1354	assert_eq!(section.file(), target);
1355	assert_same(section.page(space).unwrap(), &edited(&before, "Moved "));
1356	*release.0.lock().unwrap() = true;
1357	release.1.notify_all();
1358	stopped.recv_timeout(Duration::from_secs(3)).unwrap();
1359	assert_eq!(
1360	publications.load(Ordering::SeqCst),
1361	usize::from(!late_error)
1362	);
1363	assert!(section.sync_status().unwrap().error.is_none());
1364	assert!(!matches!(
1365	section.status(edit).unwrap(),
1366	Some(EditStatus::Published { .. })
1367	));
1368	section.set_offline(false);
1369	published(&section, edit);
1370	assert_same(stored_page(&target, space), &edited(&before, "Moved "));
1371	assert!(!file.exists());
1372	section.close().unwrap();
1373	}
1374	}
1375
1376	impl notebook::Remote for Counted {
1377	fn read(&mut self) -> std::io::Result<Vec<u8>> {
1378	onestore::read_file(&self.0)
1379	}
1380
1381	fn stamp(&mut self) -> std::io::Result<onestore::Stamp> {
1382	self.1.fetch_add(1, Ordering::SeqCst);
1383	onestore::Stamp::of(&self.read()?).map_err(std::io::Error::other)
1384	}
1385
1386	fn publish(
1387	&mut self,
1388	transaction: &onestore::Transaction,
1389	) -> Result<(), onestore::CommitError> {
1390	transaction.commit_file(&self.0)
1391	}
1392
1393	fn confirm(&mut self, base: &onestore::Stamp) -> Result<(), onestore::CommitError> {
1394	onestore::confirm_file(&self.0, base)
1395	}
1396	}
1397
1398	#[test]
1399	fn an_open_section_the_background_holds_checks_its_file_when_the_watch_reports_it() {
1400	let directory = tempfile::tempdir().unwrap();
1401	let mut notebook = two_sections(directory.path());
1402	let background = notebook.background(true, false, || {}).unwrap();
1403	background.watch(notebook.replicas());
1404	until("both sections were reached", || {
1405	background
1406	.status()
1407	.iter()
1408	.all(|(_, status)| status.synced.is_some())
1409	});
1410	let stamps = Arc::new(AtomicUsize::new(0));
1411	let counter = Arc::clone(&stamps);
1412	let section = notebook
1413	.section_with(
1414	"Second.one",
1415	move |file| Ok(Counted(file.to_owned(), Arc::clone(&counter))),
1416	false,
1417	|| {},
1418	)
1419	.unwrap();
1420	background.hold("Second.one", &section);
1421	std::thread::sleep(Duration::from_secs(3));
1422	let before = stamps.load(Ordering::SeqCst);
1423	std::thread::sleep(Duration::from_secs(5));
1424	assert_eq!(
1425	stamps.load(Ordering::SeqCst),
1426	before,
1427	"an idle held section reads nothing"
1428	);
1429
1430	let file = directory.path().join("Shared/Second.one");
1431	let bytes = onestore::read_file(&file).unwrap();
1432	let space = notebook::session::stored_pages(&bytes).unwrap()[0].space;
1433	let native = edited(&model_ops::page_of(&bytes, space), "Native ");
1434	ops::save(&bytes, space, &native)
1435	.unwrap()
1436	.commit_file(&file)
1437	.unwrap();
1438	background.touched(&["Second.one".to_owned()]);
1439	let mut changed = Vec::new();
1440	until("the session heard of the change", || {
1441	changed.extend(
1442	section
1443	.events()
1444	.into_iter()
1445	.filter_map(|event| match event {
1446	Event::Changed(spaces) => Some(spaces),
1447	_ => None,
1448	}),
1449	);
1450	!changed.is_empty()
1451	});
1452	assert!(changed.concat().contains(&space));
1453	assert!(background.changed().is_empty(), "the session reports it");
1454
1455	// Once the session stops, the background takes the section over at once.
1456	section.set_offline(true);
1457	let before = section.page(space).unwrap();
1458	typed(&section, space, &before, 0..0, "Closed ");
1459	section.close().unwrap();
1460	let mut after = edited(&before, "Closed ");
1461	until("the background published the closed section's edit", || {
1462	let stored = stored_page(&file, space);
1463	after.title = stored.title.clone();
1464	stored == after
1465	});
1466	}
1467
1468	/// A section file whose whole reads are counted.
1469	struct Reads(std::path::PathBuf, Arc<AtomicUsize>);
1470
1471	impl notebook::Remote for Reads {
1472	fn read(&mut self) -> std::io::Result<Vec<u8>> {
1473	self.1.fetch_add(1, Ordering::SeqCst);
1474	onestore::read_file(&self.0)
1475	}
1476
1477	fn stamp(&mut self) -> std::io::Result<onestore::Stamp> {
1478	onestore::Stamp::of(&onestore::read_file(&self.0)?).map_err(std::io::Error::other)
1479	}
1480
1481	fn publish(
1482	&mut self,
1483	transaction: &onestore::Transaction,
1484	) -> Result<(), onestore::CommitError> {
1485	transaction.commit_file(&self.0)
1486	}
1487
1488	fn confirm(&mut self, base: &onestore::Stamp) -> Result<(), onestore::CommitError> {
1489	onestore::confirm_file(&self.0, base)
1490	}
1491	}
1492
1493	#[test]
1494	fn the_background_takes_each_file_discovery_read_rather_than_reading_it_again() {
1495	let directory = tempfile::tempdir().unwrap();
1496	let mut notebook = two_sections(directory.path());
1497	let reads = Arc::new(AtomicUsize::new(0));
1498	let counter = Arc::clone(&reads);
1499	let remote = move |file: &Path| Reads(file.to_owned(), Arc::clone(&counter));
1500	let synced = |background: &notebook::session::Background| {
1501	let status = background.status();
1502	status.len() == 2 && status.iter().all(|(_, status)| status.synced.is_some())
1503	};
1504	// A cold launch makes each offline copy from the file as discovery read it.
1505	let background = notebook
1506	.background_with(true, true, remote.clone(), || {})
1507	.unwrap();
1508	background.watch(notebook.replicas());
1509	until("both sections have offline copies", || synced(&background));
1510	assert_eq!(reads.load(Ordering::SeqCst), 0);
1511	drop(background);
1512	let replica = notebook.replica_path("First.one").unwrap();
1513	drop(notebook);
1514
1515	// Another client changes a section while the notebook is closed.
1516	let root = directory.path().join("Shared");
1517	let file = root.join("First.one");
1518	let bytes = onestore::read_file(&file).unwrap();
1519	let space = notebook::session::stored_pages(&bytes).unwrap()[0].space;
1520	let native = edited(&model_ops::page_of(&bytes, space), "Native ");
1521	ops::save(&bytes, space, &native)
1522	.unwrap()
1523	.commit_file(&file)
1524	.unwrap();
1525	let mut notebook = Notebook::open(&root, directory.path().join("cache")).unwrap();
1526	let background = notebook.background_with(true, true, remote, || {}).unwrap();
1527	background.watch(notebook.replicas());
1528	until("both sections were checked", || synced(&background));
1529	assert_eq!(
1530	reads.load(Ordering::SeqCst),
1531	0,
1532	"the rebase takes the image discovery read"
1533	);
1534	drop(background);
1535	let rebased = notebook::Replica::open(&replica).unwrap();
1536	assert_eq!(rebased.page(space).unwrap(), stored_page(&file, space));
1537	}
1538
1539	/// Every file under `dir`, depth first.
1540	fn walkdir(dir: &std::path::Path) -> Vec<std::path::PathBuf> {
1541	let mut files = Vec::new();
1542	for entry in std::fs::read_dir(dir).unwrap() {
1543	let path = entry.unwrap().path();
1544	if path.is_dir() {
1545	files.extend(walkdir(&path));
1546	} else {
1547	files.push(path);
1548	}
1549	}
1550	files
1551	}
1552
1553	/// A stored page names who changed it last, which tells another author's changes from
1554	/// one's own.
1555	#[test]
1556	fn a_stored_page_names_who_changed_it_last() {
1557	let image = onestore::create_section("Authors.one", "Body", "Alice").unwrap();
1558	let first = &notebook::session::stored_pages(&image).unwrap()[0];
1559	assert_eq!(first.author.as_deref(), Some("Alice"));
1560	let arena = onestore::Arena::default();
1561	let mut section = onestore::Section::open(&arena, image).unwrap();
1562	// Enter, which stamps both paragraphs with who pressed it.
1563	let new_id = || onestore::page::text::new_id().unwrap();
1564	let op = PageOp::Split {
1565	text: first_text(&first.page),
1566	at: 0,
1567	paragraph: new_id(),
1568	right: new_id(),
1569	lists: Vec::new(),
1570	};
1571	let edit = Edit {
1572	// A minute on, so the edit is the page's latest change.
1573	at: model_ops::now() + 600_000_000,
1574	ops: vec![Op::Page {
1575	space: first.space,
1576	op,
1577	}],
1578	};
1579	section.apply("Bob", &edit).unwrap();
1580	section.seal().unwrap();
1581	let edited = &notebook::session::stored_pages(&section.image()).unwrap()[0];
1582	assert_eq!(edited.author.as_deref(), Some("Bob"));
1583	}
1584
1585	#[test]
1586	fn text_edits_name_the_latest_author_even_within_the_same_second() {
1587	let image = onestore::create_section("Authors.one", "Body", "Alice").unwrap();
1588	let first = notebook::session::stored_pages(&image).unwrap().remove(0);
1589	let at = first.modified.unwrap() + 12345;
1590	let arena = onestore::Arena::default();
1591	let mut section = onestore::Section::open(&arena, image).unwrap();
1592	for (author, at) in [("Bob", at), ("Alice", at + 1), ("Bob", at + 20_000_000)] {
1593	let before = section.image();
1594	let store = onestore::Store::parse(&before).unwrap();
1595	let index = onestore::RevisionIndex::parse(&store).unwrap();
1596	let revisions = index.spaces[&first.space].revisions.len();
1597	let edit = Edit {
1598	at,
1599	ops: vec![Op::Page {
1600	space: first.space,
1601	op: PageOp::Text {
1602	text: first_text(&first.page),
1603	range: 0..0,
1604	with: author.into(),
1605	},
1606	}],
1607	};
1608	section.apply(author, &edit).unwrap();
1609	section.seal().unwrap().unwrap();
1610	let image = section.image();
1611	let page = notebook::session::stored_pages(&image).unwrap().remove(0);
1612	assert_eq!(page.author.as_deref(), Some(author));
1613	assert_eq!(page.modified, Some(at));
1614	let store = onestore::Store::parse(&image).unwrap();
1615	let index = onestore::RevisionIndex::parse(&store).unwrap();
1616	assert_eq!(index.spaces[&first.space].revisions.len(), revisions + 1);
1617	let document = onestore::document::Document::parse(&index).unwrap();
1618	let revision = document.active(first.space).unwrap();
1619	let metadata = &revision.nodes[&revision.roots[&4]];
1620	assert!(
1621	matches!(metadata.kind, onestore::document::Kind::RevisionMetadata { modified_filetime: Some(stored) } if stored == at)
1622	);
1623	let paragraph = revision
1624	.nodes
1625	.values()
1626	.find(|node| node.content.contains(&first_text(&first.page)))
1627	.unwrap();
1628	let name = |id: Option<ExGuid>| match &revision.nodes[&id.unwrap()].kind {
1629	onestore::document::Kind::Author { name } => name.clone().unwrap(),
1630	_ => panic!("author reference names another kind of object"),
1631	};
1632	assert_eq!(name(paragraph.original_author), "Alice");
1633	assert_eq!(name(paragraph.latest_author), author);
1634	}
1635	section
1636	.apply(
1637	"Nobody",
1638	&Edit {
1639	at: at + 30_000_000,
1640	ops: vec![Op::Page {
1641	space: first.space,
1642	op: PageOp::Text {
1643	text: first_text(&first.page),
1644	range: 0..0,
1645	with: String::new(),
1646	},
1647	}],
1648	},
1649	)
1650	.unwrap();
1651	assert!(section.seal().unwrap().is_none());
1652	if let Some(folder) = std::env::var_os("SNOWBOUND_AUTHOR_EXPORT") {
1653	std::fs::create_dir(&folder).unwrap();
1654	std::fs::write(
1655	std::path::Path::new(&folder).join("Authors.one"),
1656	section.image(),
1657	)
1658	.unwrap();
1659	}
1660	}
1661
1662	#[test]
1663	fn page_level_edits_attribute_only_changed_pages_and_skip_noops() {
1664	let image = onestore::create_section("Levels.one", "Parent", "Alice").unwrap();
1665	let arena = onestore::Arena::default();
1666	let mut section = onestore::Section::open(&arena, image).unwrap();
1667	let creation = onestore::PageCreation::new(None, Some("Child"), "Alice").unwrap();
1668	let child = creation.space();
1669	section
1670	.apply(
1671	"Alice",
1672	&Edit {
1673	at: model_ops::now(),
1674	ops: vec![Op::Section(onestore::op::SectionOp::Create(creation))],
1675	},
1676	)
1677	.unwrap();
1678	section.seal().unwrap().unwrap();
1679	let before = section.image();
1680	let pages = notebook::session::stored_pages(&before).unwrap();
1681	let at = pages.iter().filter_map(|page| page.modified).max().unwrap() + 12345;
1682	let store = onestore::Store::parse(&before).unwrap();
1683	let index = onestore::RevisionIndex::parse(&store).unwrap();
1684	let revisions = index.spaces[&child].revisions.len();
1685	let edit = Edit {
1686	at,
1687	ops: vec![Op::Section(onestore::op::SectionOp::Pages(vec![
1688	onestore::PageEdit::set_level(child, 2).unwrap(),
1689	]))],
1690	};
1691	section.apply("Bob", &edit).unwrap();
1692	section.seal().unwrap().unwrap();
1693	let image = section.image();
1694	let after = notebook::session::stored_pages(&image).unwrap();
1695	assert_eq!(after.len(), 2);
1696	for (before, after) in pages.iter().zip(&after) {
1697	if after.space == child {
1698	assert_eq!(after.author.as_deref(), Some("Bob"));
1699	assert_eq!(after.modified, Some(at));
1700	} else {
1701	assert_eq!(after.author, before.author);
1702	assert_eq!(after.modified, before.modified);
1703	}
1704	}
1705	assert_eq!(section.pages().unwrap()[1].2, 2);
1706	let store = onestore::Store::parse(&image).unwrap();
1707	let index = onestore::RevisionIndex::parse(&store).unwrap();
1708	assert_eq!(index.spaces[&child].revisions.len(), revisions + 1);
1709	section
1710	.apply("Nobody", &Edit { at: at + 1, ..edit })
1711	.unwrap();
1712	assert!(section.seal().unwrap().is_none());
1713	assert_eq!(section.image(), image);
1714	if let Some(folder) = std::env::var_os("SNOWBOUND_AUTHOR_EXPORT") {
1715	std::fs::create_dir_all(&folder).unwrap();
1716	std::fs::write(std::path::Path::new(&folder).join("Levels.one"), image).unwrap();
1717	}
1718	}