clover's git

snowbound

1	use super::*;
2
3	#[cfg(unix)]
4	#[test]
5	fn cached_sections_refuse_links_outside_the_notebook_and_dangling_links() {
6	let temporary = tempfile::tempdir().unwrap();
7	let folder = temporary.path().join("notebook");
8	let cache = temporary.path().join("cache");
9	let creation = PageCreation::new(None, Some("Mine"), "Fixture").unwrap();
10	let notebook = Notebook::create(&folder, &cache, 0x00f0_a0c0, &creation).unwrap();
11	let path = &notebook.catalog().sections[0].path;
12	notebook
13	.section(path, false, || {})
14	.unwrap()
15	.close()
16	.unwrap();
17	let file = folder.join(path);
18	let outside = temporary.path().join("outside.one");
19	fs::rename(&file, &outside).unwrap();
20	std::os::unix::fs::symlink(&outside, &file).unwrap();
21	assert!(notebook.section(path, false, || {}).is_err());
22	fs::remove_file(outside).unwrap();
23	assert!(notebook.section(path, false, || {}).is_err());
24	}
25
26	#[test]
27	fn renaming_an_unlisted_section_creates_its_toc_entry_and_preserves_its_neighbor() {
28	let temporary = tempfile::tempdir().unwrap();
29	let folder = temporary.path().join("notebook");
30	let cache = temporary.path().join("cache");
31	let page = PageCreation::new(None, Some("Garden plan"), "Fixture").unwrap();
32	let mut notebook = Notebook::create(&folder, &cache, Notebook::NEW_COLOR, &page).unwrap();
33	let original = notebook.catalog().sections[0].path.clone();
34	let other = notebook.create_section("", "Other", &page).unwrap();
35	let identity = notebook.section_path(&original).unwrap().file_id;
36	let neighbor = notebook.section_path(&other).unwrap().file_id;
37	drop(notebook);
38	fs::remove_file(folder.join(TOC)).unwrap();
39	let mut notebook = Notebook::open(&folder, &cache).unwrap();
40	assert!(notebook.catalog().toc.is_none());
41	assert_eq!(notebook.catalog().sections.len(), 2);
42	assert_eq!(
43	notebook.rename(&original, "Delayed section").unwrap(),
44	"Delayed section.one"
45	);
46	assert!(!folder.join(&original).exists());
47	let check = |notebook: &Notebook| {
48	assert_eq!(notebook.catalog().sections.len(), 2);
49	assert!(notebook.section_path(&original).is_err());
50	assert_eq!(
51	notebook
52	.section_path("Delayed section.one")
53	.unwrap()
54	.file_id,
55	identity
56	);
57	assert_eq!(notebook.section_path(&other).unwrap().file_id, neighbor);
58	let toc = notebook.read_toc("").unwrap();
59	assert!(lists(&toc, identity).unwrap());
60	assert_eq!(
61	stored_pages(&notebook.read_section("Delayed section.one").unwrap()).unwrap()[0]
62	.page
63	.title,
64	"Garden plan"
65	);
66	};
67	check(&notebook);
68	drop(notebook);
69	check(&Notebook::open(&folder, temporary.path().join("cold-cache")).unwrap());
70	if let Some(export) = std::env::var_os("SNOWBOUND_RENAME_EXPORT") {
71	let export = Path::new(&export);
72	fs::create_dir_all(export).unwrap();
73	for entry in fs::read_dir(&folder).unwrap() {
74	let entry = entry.unwrap();
75	fs::copy(entry.path(), export.join(entry.file_name())).unwrap();
76	}
77	}
78	}
79
80	#[test]
81	fn moving_discovered_entries_preserves_unregistered_source_tocs() {
82	for group in [false, true] {
83	for source_toc in ["missing", "empty", "unlisted", "registered"] {
84	let temporary = tempfile::tempdir().unwrap();
85	let folder = temporary.path().join("notebook");
86	let cache = temporary.path().join("cache");
87	let page = PageCreation::new(None, Some("Garden plan"), "Fixture").unwrap();
88	let mut notebook =
89	Notebook::create(&folder, &cache, Notebook::NEW_COLOR, &page).unwrap();
90	let first = notebook.catalog().sections[0].path.clone();
91	let neighbor = notebook.create_section("", "Neighbor", &page).unwrap();
92	let target = notebook.create_group("", "Target").unwrap();
93	let existing = notebook.create_section(&target, "Existing", &page).unwrap();
94	let (entry, section, identity) = if group {
95	let entry = notebook.create_group("", "Source").unwrap();
96	let section = notebook.create_section(&entry, "Inside", &page).unwrap();
97	let identity = notebook
98	.folder(&entry)
99	.unwrap()
100	.toc
101	.as_ref()
102	.unwrap()
103	.file_id;
104	(entry, section, identity)
105	} else {
106	let identity = notebook.section_path(&first).unwrap().file_id;
107	(first.clone(), first, identity)
108	};
109	let section_id = notebook.section_path(&section).unwrap().file_id;
110	let neighbor_bytes = fs::read(folder.join(&neighbor)).unwrap();
111	let existing_bytes = fs::read(folder.join(&existing)).unwrap();
112	if source_toc == "unlisted" {
113	notebook
114	.edit_toc("", &[onestore::TocEdit::Remove { identity }])
115	.unwrap();
116	}
117	drop(notebook);
118	if source_toc == "missing" {
119	fs::remove_file(folder.join(TOC)).unwrap();
120	} else if source_toc == "empty" {
121	fs::write(
122	folder.join(TOC),
123	onestore::create_table_of_contents(TOC, &[]).unwrap(),
124	)
125	.unwrap();
126	}
127	let before = fs::read(folder.join(TOC)).ok();
128	let mut notebook = Notebook::open(&folder, &cache).unwrap();
129	let order: Vec<_> = notebook
130	.catalog()
131	.order
132	.iter()
133	.filter(|path| **path != entry)
134	.cloned()
135	.collect();
136	let moved = catalog_path(&target, &entry);
137	assert_eq!(notebook.move_entry(&entry, &target).unwrap(), moved);
138	assert!(!folder.join(&entry).exists());
139	let moved_section = catalog_path(&target, &section);
140	let check = |notebook: &Notebook| {
141	assert_eq!(notebook.catalog().order, order);
142	assert_eq!(
143	notebook.section_path(&moved_section).unwrap().file_id,
144	section_id
145	);
146	assert!(lists(&notebook.read_toc(&target).unwrap(), identity).unwrap());
147	assert_eq!(
148	stored_pages(&notebook.read_section(&moved_section).unwrap()).unwrap()[0]
149	.page
150	.title,
151	"Garden plan"
152	);
153	if source_toc == "registered" {
154	let toc = notebook.read_toc("").unwrap();
155	assert!(!lists(&toc, identity).unwrap());
156	assert_eq!(
157	notebook.catalog().toc.as_ref().unwrap().color,
158	Some(Notebook::NEW_COLOR)
159	);
160	} else {
161	assert_eq!(fs::read(folder.join(TOC)).ok(), before);
162	}
163	assert_eq!(fs::read(folder.join(&neighbor)).unwrap(), neighbor_bytes);
164	assert_eq!(fs::read(folder.join(&existing)).unwrap(), existing_bytes);
165	};
166	check(&notebook);
167	drop(notebook);
168	check(&Notebook::open(&folder, temporary.path().join("cold-cache")).unwrap());
169	}
170	}
171	}
172
173	#[test]
174	fn moving_an_entry_reads_its_registered_source_toc_before_renaming() {
175	let temporary = tempfile::tempdir().unwrap();
176	let folder = temporary.path().join("notebook");
177	let page = PageCreation::new(None, Some("Garden plan"), "Fixture").unwrap();
178	let mut notebook = Notebook::create(
179	&folder,
180	temporary.path().join("cache"),
181	Notebook::NEW_COLOR,
182	&page,
183	)
184	.unwrap();
185	let section = notebook.catalog().sections[0].path.clone();
186	notebook.create_group("", "Target").unwrap();
187	fs::write(folder.join(TOC), b"unreadable source TOC").unwrap();
188	assert!(notebook.move_entry(&section, "Target").is_err());
189	assert!(folder.join(&section).exists());
190	assert!(!folder.join("Target").join(&section).exists());
191	}
192
193	struct FaultStorage {
194	inner: Directory,
195	revision: Mutex<Option<(String, Transaction)>>,
196	receipt: Option<Arc<Mutex<Receipt>>>,
197	}
198
199	#[derive(Default)]
200	struct Receipt {
201	reject: Option<String>,
202	reject_confirmation: Option<String>,
203	confirmed: Vec<String>,
204	}
205
206	impl FaultStorage {
207	fn race(&self, path: &str) {
208	let revision = self.revision.lock().unwrap().take();
209	if let Some((section, transaction)) = revision {
210	assert_eq!(section, path);
211	self.inner.commit(&section, &transaction).unwrap();
212	}
213	}
214	}
215
216	impl Storage for FaultStorage {
217	fn discover(
218	&self,
219	cache: &mut discover::Cache,
220	limits: discover::Limits,
221	) -> Result<discover::Folder> {
222	self.inner.discover(cache, limits)
223	}
224	fn location(&self) -> String {
225	self.inner.location()
226	}
227	fn entries(&self, folder: &str) -> io::Result<Vec<discover::Entry>> {
228	self.inner.entries(folder)
229	}
230	fn exists(&self, path: &str) -> bool {
231	self.inner.exists(path)
232	}
233	fn stamp(&self, path: &str) -> io::Result<Stamp> {
234	self.inner.stamp(path)
235	}
236	fn read(&self, path: &str) -> Result<Vec<u8>> {
237	self.inner.read(path)
238	}
239	fn read_file(&self, path: &str, limit: usize) -> Result<Vec<u8>> {
240	self.inner.read_file(path, limit)
241	}
242	fn create(&self, path: &str, bytes: &[u8]) -> Result<()> {
243	self.inner.create(path, bytes)
244	}
245	fn create_directory(&self, path: &str) -> Result<()> {
246	self.inner.create_directory(path)
247	}
248	fn hide(&self, path: &str) -> Result<()> {
249	self.inner.hide(path)
250	}
251	fn rename(&self, from: &str, to: &str) -> Result<()> {
252	self.inner.rename(from, to)
253	}
254	fn rename_root(&self, name: &str, files: &[String]) -> Result<String> {
255	self.inner.rename_root(name, files)
256	}
257	fn replace(&self, from: &str, to: &str) -> Result<()> {
258	self.inner.replace(from, to)
259	}
260	fn delete(&self, path: &str) -> Result<()> {
261	self.inner.delete(path)
262	}
263	fn place(&self, path: &str, ancestor: [u8; 16], name: &str) -> Result<()> {
264	self.inner.place(path, ancestor, name)
265	}
266	fn commit(&self, path: &str, transaction: &Transaction) -> Result<()> {
267	if let Some(receipt) = &self.receipt {
268	if receipt.lock().unwrap().reject.as_deref() != Some(path) {
269	self.inner.commit(path, transaction)?;
270	}
271	return Err(CommitError {
272	state: onestore::CommitState::Unknown,
273	error: io::ErrorKind::WouldBlock.into(),
274	}
275	.into());
276	}
277	self.inner.commit(path, transaction)
278	}
279	fn confirm(&self, path: &str, base: &Stamp) -> std::result::Result<(), CommitError> {
280	if let Some(receipt) = &self.receipt {
281	let mut receipt = receipt.lock().unwrap();
282	receipt.confirmed.push(path.into());
283	if receipt.reject_confirmation.as_deref() == Some(path) {
284	return Err(CommitError {
285	state: onestore::CommitState::Unknown,
286	error: io::ErrorKind::PermissionDenied.into(),
287	});
288	}
289	}
290	self.inner.confirm(path, base)
291	}
292	fn supersede(&self, path: &str, base: &Stamp, with: &str) -> Result<()> {
293	self.race(path);
294	self.inner.supersede(path, base, with)
295	}
296	}
297
298	/// A revision another writer commits while a password is being set is never overwritten:
299	/// the section keeps it, and the password waits for a later try.
300	#[test]
301	fn a_password_never_overwrites_a_concurrent_revision() {
302	let temporary = tempfile::tempdir().unwrap();
303	let folder = temporary.path().join("notebook");
304	let cache = temporary.path().join("cache");
305	let creation = PageCreation::new(None, Some("Mine"), "Fixture").unwrap();
306	let path = Notebook::create(&folder, &cache, 0x00f0_a0c0, &creation)
307	.unwrap()
308	.catalog()
309	.sections[0]
310	.path
311	.clone();
312	let file = folder.canonicalize().unwrap().join(&path);
313	let image = std::fs::read(&file).unwrap();
314	let arena = onestore::Arena::default();
315	let mut section = onestore::Section::open(&arena, image).unwrap();
316	let theirs = PageCreation::new(None, Some("Theirs"), "OneNote").unwrap();
317	section
318	.apply(
319	"OneNote",
320	&Edit {
321	at: crate::now(),
322	ops: vec![Op::Section(SectionOp::Create(theirs))],
323	},
324	)
325	.unwrap();
326	let revision = section.seal().unwrap().unwrap();
327
328	let root = folder.canonicalize().unwrap();
329	let storage = FaultStorage {
330	inner: Directory(root.clone()),
331	revision: Mutex::new(Some((path.clone(), revision))),
332	receipt: None,
333	};
334	let mut notebook = Notebook::with(Box::new(storage), Some(root), &cache).unwrap();
335	let set = notebook.set_password(&path, None, Some("secret"));
336	let titles = |image: &[u8]| -> Vec<String> {
337	stored_pages(image)
338	.unwrap()
339	.into_iter()
340	.map(|page| page.page.title)
341	.collect()
342	};
343	let stored = std::fs::read(&file).unwrap();
344	match set {
345	Ok(key) => {
346	let key = key.unwrap();
347	let pages = stored_pages_unlocked(&stored, &key).unwrap();
348	assert!(pages.iter().any(|page| page.page.title == "Theirs"));
349	}
350	Err(_) => assert_eq!(titles(&stored), ["Mine", "Theirs"]),
351	}
352	}
353
354	#[test]
355	fn recycling_confirms_uncertain_toc_and_page_appends_and_undo() {
356	let temporary = tempfile::tempdir().unwrap();
357	let folder = temporary.path().join("notebook");
358	let cache = temporary.path().join("cache");
359	let creation = PageCreation::new(None, Some("Keep this page"), "Fixture").unwrap();
360	let notebook = Notebook::create(&folder, &cache, Notebook::NEW_COLOR, &creation).unwrap();
361	let path = notebook.catalog().sections[0].path.clone();
362	let source = notebook.read_section(&path).unwrap();
363	let page = stored_pages(&source).unwrap().remove(0).page;
364	drop(notebook);
365	let root = folder.canonicalize().unwrap();
366	let receipt = Arc::new(Mutex::new(Receipt::default()));
367	let storage = FaultStorage {
368	inner: Directory(root.clone()),
369	revision: Mutex::new(None),
370	receipt: Some(receipt.clone()),
371	};
372	let mut notebook = Notebook::with(Box::new(storage), Some(root), &cache).unwrap();
373	notebook
374	.recycle_pages(std::slice::from_ref(&page), "Fixture")
375	.unwrap();
376	let deleted = catalog_path(RECYCLE_BIN, "OneNote_DeletedPages.one");
377	assert_eq!(
378	receipt.lock().unwrap().confirmed,
379	[
380	TOC.to_string(),
381	catalog_path(RECYCLE_BIN, TOC),
382	deleted.clone(),
383	]
384	);
385	let binned = fs::read(folder.join(&deleted)).unwrap();
386	let pages = stored_pages(&binned).unwrap();
387	assert_eq!(pages.len(), 1);
388	assert_eq!(pages[0].page.title, page.title);
389	assert_eq!(pages[0].page.identity, page.identity);
390	assert_eq!(pages[0].page.created, page.created);
391	assert_eq!(pages[0].page.date_text(), page.date_text());
392	let cold = Notebook::open(&folder, temporary.path().join("cold-cache")).unwrap();
393	assert_eq!(cold.catalog().groups.len(), 1);
394	assert_eq!(cold.folder(RECYCLE_BIN).unwrap().sections.len(), 1);
395	assert_eq!(
396	stored_pages(&cold.read_section(&deleted).unwrap())
397	.unwrap()
398	.len(),
399	1
400	);
401	drop(cold);
402	notebook.unrecycle_pages(&[page.identity.unwrap()]).unwrap();
403	let restored = fs::read(folder.join(&deleted)).unwrap();
404	assert!(
405	restored.len() > binned.len(),
406	"Undo must append its deletion revision"
407	);
408	assert!(stored_pages(&restored).unwrap().is_empty());
409	assert_eq!(receipt.lock().unwrap().confirmed.last(), Some(&deleted));
410	assert_eq!(receipt.lock().unwrap().confirmed.len(), 4);
411	assert_eq!(fs::read(folder.join(&path)).unwrap(), source);
412	}
413
414	#[test]
415	fn recycling_rejects_unconfirmed_appends_and_can_retry_each_stage() {
416	let deleted = catalog_path(RECYCLE_BIN, "OneNote_DeletedPages.one");
417	for (rejected, retained) in [
418	(TOC.to_string(), false),
419	(catalog_path(RECYCLE_BIN, TOC), false),
420	(deleted.clone(), false),
421	(deleted.clone(), true),
422	] {
423	let temporary = tempfile::tempdir().unwrap();
424	let folder = temporary.path().join("notebook");
425	let cache = temporary.path().join("cache");
426	let creation = PageCreation::new(None, Some("Keep this page"), "Fixture").unwrap();
427	let notebook = Notebook::create(&folder, &cache, Notebook::NEW_COLOR, &creation).unwrap();
428	let path = notebook.catalog().sections[0].path.clone();
429	let mut source = notebook.read_section(&path).unwrap();
430	let mut page = stored_pages(&source).unwrap().remove(0).page;
431	drop(notebook);
432	let root = folder.canonicalize().unwrap();
433	let receipt = Arc::new(Mutex::new(Receipt {
434	reject: (!retained).then(|| rejected.clone()),
435	reject_confirmation: retained.then(|| rejected.clone()),
436	confirmed: Vec::new(),
437	}));
438	let storage = FaultStorage {
439	inner: Directory(root.clone()),
440	revision: Mutex::new(None),
441	receipt: Some(receipt.clone()),
442	};
443	let mut notebook = Notebook::with(Box::new(storage), Some(root), &cache).unwrap();
444	assert!(
445	matches!(
446	notebook.recycle_pages(std::slice::from_ref(&page), "Fixture"),
447	Err(Error::Remote(_))
448	),
449	"{rejected}"
450	);
451	assert_eq!(receipt.lock().unwrap().confirmed.last(), Some(&rejected));
452	assert_eq!(fs::read(folder.join(&path)).unwrap(), source);
453	if folder.join(&deleted).exists() {
454	assert_eq!(
455	stored_pages(&fs::read(folder.join(&deleted)).unwrap())
456	.unwrap()
457	.len(),
458	usize::from(retained)
459	);
460	}
461	if retained {
462	let arena = onestore::Arena::default();
463	let mut section = onestore::Section::open(&arena, source).unwrap();
464	let space = section.pages().unwrap()[0].0;
465	section
466	.apply(
467	"Fixture",
468	&Edit {
469	at: crate::now(),
470	ops: vec![Op::Page {
471	space,
472	op: onestore::op::PageOp::Color(Some(0xabcdef)),
473	}],
474	},
475	)
476	.unwrap();
477	section
478	.seal()
479	.unwrap()
480	.unwrap()
481	.commit_file(folder.join(&path))
482	.unwrap();
483	source = fs::read(folder.join(&path)).unwrap();
484	page = stored_pages(&source).unwrap().remove(0).page;
485	}
486	{
487	let mut receipt = receipt.lock().unwrap();
488	receipt.reject = None;
489	receipt.reject_confirmation = None;
490	}
491	notebook
492	.recycle_pages(std::slice::from_ref(&page), "Fixture")
493	.unwrap();
494	let binned = stored_pages(&fs::read(folder.join(&deleted)).unwrap()).unwrap();
495	assert_eq!(binned.len(), 1 + usize::from(retained), "{rejected}");
496	for snapshot in &binned {
497	assert_eq!(snapshot.page.title, page.title);
498	assert_eq!(snapshot.page.identity, page.identity);
499	}
500	assert_eq!(binned.last().unwrap().page.color, page.color);
501	if retained {
502	assert_eq!(binned[0].page.color, None);
503	assert_eq!(binned.last().unwrap().page.color, Some(0xabcdef));
504	}
505	assert_eq!(fs::read(folder.join(&path)).unwrap(), source);
506	}
507	}
508
509	#[test]
510	fn initially_offline_workers_keep_the_original_queue_until_explicitly_woken() {
511	use std::sync::atomic::{AtomicUsize, Ordering};
512	use std::time::Instant;
513
514	struct Counted {
515	file: FileRemote,
516	calls: Arc<[AtomicUsize; 5]>,
517	}
518	impl Remote for Counted {
519	fn read(&mut self) -> io::Result<Vec<u8>> {
520	self.calls[1].fetch_add(1, Ordering::SeqCst);
521	self.file.read()
522	}
523	fn stamp(&mut self) -> io::Result<Stamp> {
524	self.calls[2].fetch_add(1, Ordering::SeqCst);
525	self.file.stamp()
526	}
527	fn publish(&mut self, transaction: &Transaction) -> std::result::Result<(), CommitError> {
528	self.calls[3].fetch_add(1, Ordering::SeqCst);
529	self.file.publish(transaction)
530	}
531	fn confirm(&mut self, base: &Stamp) -> std::result::Result<(), CommitError> {
532	self.calls[4].fetch_add(1, Ordering::SeqCst);
533	self.file.confirm(base)
534	}
535	}
536
537	for relocate in [false, true] {
538	let temporary = tempfile::tempdir().unwrap();
539	let file = temporary.path().join("Original.one");
540	let cache = temporary.path().join("original.sqlite");
541	let source = onestore::create_section("Original.one", "Original", "Fixture").unwrap();
542	fs::write(&file, &source).unwrap();
543	let replica = Replica::create(&cache, &source).unwrap();
544	let creation =
545	PageCreation::new(None, Some("Offline initialization marker"), "Fixture").unwrap();
546	let id = replica
547	.apply(
548	"Fixture",
549	Edit {
550	at: crate::now(),
551	ops: vec![Op::Section(SectionOp::Create(creation))],
552	},
553	)
554	.unwrap();
555	let pending = replica.pending().unwrap();
556	drop(replica);
557	let calls = Arc::new(std::array::from_fn(|_| AtomicUsize::new(0)));
558	let counter = Arc::clone(&calls);
559	let remote = file.clone();
560	let (notify, notified) = mpsc::channel();
561	let mut section = Section::start(
562	file.clone(),
563	Replica::open(&cache).unwrap(),
564	move || {
565	counter[0].fetch_add(1, Ordering::SeqCst);
566	Ok(Counted {
567	file: FileRemote(remote.clone()),
568	calls: Arc::clone(&counter),
569	})
570	},
571	CHECK,
572	true,
573	move || {
574	let _ = notify.send(());
575	},
576	)
577	.unwrap();
578	let target = if relocate {
579	let target = temporary.path().join("Moved.one");
580	fs::rename(&file, &target).unwrap();
581	let counter = Arc::clone(&calls);
582	section
583	.relocate_with(&target, move |file| {
584	counter[0].fetch_add(1, Ordering::SeqCst);
585	Ok(Counted {
586	file: FileRemote(file.to_owned()),
587	calls: Arc::clone(&counter),
588	})
589	})
590	.unwrap();
591	target
592	} else {
593	file
594	};
595	let deadline = Instant::now() + Duration::from_millis(300);
596	while Instant::now() < deadline {
597	assert_eq!(section.pending().unwrap(), pending);
598	assert_eq!(fs::read(&target).unwrap(), source);
599	assert!(calls.iter().all(|count| count.load(Ordering::SeqCst) == 0));
600	std::thread::sleep(Duration::from_millis(10));
601	}
602	section.wake();
603	let deadline = Instant::now() + Duration::from_secs(10);
604	loop {
605	notified
606	.recv_timeout(deadline.saturating_duration_since(Instant::now()))
607	.unwrap();
608	if section.events().iter().any(|event| matches!(event,
609	Event::Attempt { id: published, status: EditStatus::Published { .. } } if *published == id)) {
610	break;
611	}
612	}
613	assert!(section.pending().unwrap().is_empty());
614	assert_eq!(calls[0].load(Ordering::SeqCst), 1);
615	assert_eq!(calls[3].load(Ordering::SeqCst), 1);
616	section.close().unwrap();
617	let pages = stored_pages(&fs::read(&target).unwrap()).unwrap();
618	assert_eq!(
619	pages
620	.iter()
621	.filter(|page| page.page.title == "Offline initialization marker")
622	.count(),
623	1
624	);
625	assert!(Replica::open(&cache).unwrap().pending().unwrap().is_empty());
626	}
627	}
628
629	#[test]
630	fn associated_apply_reply_precedes_success_wake_without_a_publication_event() {
631	use std::sync::atomic::{AtomicBool, Ordering};
632
633	let temporary = tempfile::tempdir().unwrap();
634	let file = temporary.path().join("Reply.one");
635	let source = onestore::create_section("Reply.one", "Original", "Fixture").unwrap();
636	fs::write(&file, &source).unwrap();
637	let (woken, wake) = mpsc::channel();
638	let completed = Arc::new(AtomicBool::new(false));
639	let observed = Arc::clone(&completed);
640	let section = Section::open(&file, temporary.path().join("cache"), true, move || {
641	let _ = woken.send(observed.load(Ordering::Acquire));
642	})
643	.unwrap();
644	while wake.try_recv().is_ok() {}
645	let creation = PageCreation::new(None, Some("Associated original"), "Fixture").unwrap();
646	let space = creation.space();
647	let edit = Edit {
648	at: crate::now(),
649	ops: vec![Op::Section(SectionOp::Create(creation))],
650	};
651	let original = edit.clone();
652	let (replied, reply) = mpsc::channel();
653	section
654	.apply_with("Original author", edit, move |result| {
655	completed.store(true, Ordering::Release);
656	replied.send(result).unwrap();
657	})
658	.unwrap();
659	let id = reply
660	.recv_timeout(Duration::from_secs(10))
661	.unwrap()
662	.unwrap();
663	assert!(wake.recv_timeout(Duration::from_secs(10)).unwrap());
664	let pending = section.pending().unwrap();
665	assert_eq!(pending.len(), 1);
666	assert_eq!(pending[0].id, id);
667	assert_eq!(pending[0].author, "Original author");
668	assert_eq!(pending[0].edit, original);
669	assert_eq!(section.page(space).unwrap().title, "Associated original");
670	assert!(section.events().is_empty());
671	section.close().unwrap();
672	assert_eq!(fs::read(file).unwrap(), source);
673	}
674
675	#[test]
676	fn edits_the_cache_refuses_wait_in_order_and_land_once() {
677	use std::time::Instant;
678
679	let temporary = tempfile::tempdir().unwrap();
680	let file = temporary.path().join("Waiting.one");
681	let cache = temporary.path().join("waiting.sqlite");
682	let source = onestore::create_section("Waiting.one", "Original", "Fixture").unwrap();
683	fs::write(&file, &source).unwrap();
684	let replica = Replica::create(&cache, &source).unwrap();
685	let section = Section::resume(&file, replica, true, || {}).unwrap();
686	let replica = Arc::clone(section.replica());
687	// Queued before the fault, so the worker asks to seal while edits wait.
688	let seed = replica
689	.apply(
690	"Fixture",
691	Edit {
692	at: crate::now(),
693	ops: vec![Op::Section(SectionOp::Create(
694	PageCreation::new(None, Some("Seed"), "Fixture").unwrap(),
695	))],
696	},
697	)
698	.unwrap();
699	// The cache is full, until it may grow again; the first edit needs room.
700	let full = |pages: Option<i64>| {
701	let connection = replica.lock().unwrap();
702	let pages = pages.unwrap_or_else(|| {
703	connection
704	.pragma_query_value(None, "page_count", |row| row.get(0))
705	.unwrap()
706	});
707	connection
708	.pragma_update(None, "max_page_count", pages)
709	.unwrap();
710	};
711	full(None);
712	let title = "Waiting page ".repeat(1000);
713	let creation = PageCreation::new(None, Some(&title), "Fixture").unwrap();
714	let space = creation.space();
715	// The second edit colours the page the first creates.
716	let edits = [
717	Op::Section(SectionOp::Create(creation)),
718	Op::Page {
719	space,
720	op: onestore::op::PageOp::Color(Some(0x00ab_cdef)),
721	},
722	]
723	.map(|op| Edit {
724	at: crate::now(),
725	ops: vec![op],
726	});
727	let (replied, replies) = mpsc::channel();
728	for edit in edits.clone() {
729	let replied = replied.clone();
730	section
731	.apply_with("Fixture", edit, move |result| {
732	replied
733	.send(result.map_err(|error| error.to_string()))
734	.unwrap();
735	})
736	.unwrap();
737	}
738	let deadline = Instant::now() + Duration::from_secs(10);
739	while !section
740	.events()
741	.iter()
742	.any(|event| matches!(event, Event::Failed(_)))
743	{
744	assert!(Instant::now() < deadline);
745	std::thread::sleep(Duration::from_millis(10));
746	}
747	let waiting = |section: &Section| {
748	assert!(replies.try_recv().is_err());
749	let pending = section.pending().unwrap();
750	assert_eq!(
751	pending.iter().map(|edit| edit.id).collect::<Vec<_>>(),
752	[seed]
753	);
754	let page = section.page(space).unwrap();
755	assert_eq!((&page.title, page.color), (&title, Some(0x00ab_cdef)));
756	assert!(matches!(
757	replica.ask(|reply| crate::working::Request::Seal { reply }),
758	Err(Error::Io(error)) if error.kind() == io::ErrorKind::WouldBlock
759	));
760	assert!(matches!(
761	section.written(),
762	Err(Error::Io(error)) if error.kind() == io::ErrorKind::WouldBlock
763	));
764	};
765	waiting(&section);
766	// A caller blocking on its edit still learns of the failure, and the waiting edits stay.
767	let blocked = Edit {
768	at: crate::now(),
769	ops: vec![Op::Page {
770	space,
771	op: onestore::op::PageOp::Color(Some(0x0012_3456)),
772	}],
773	};
774	assert!(replica.apply("Fixture", blocked).is_err());
775	waiting(&section);
776	section.wake();
777	std::thread::sleep(Duration::from_millis(200));
778	waiting(&section);
779
780	full(Some(1 << 30));
781	let ids = [(); 2].map(|()| {
782	replies
783	.recv_timeout(Duration::from_secs(10))
784	.unwrap()
785	.unwrap()
786	});
787	let pending = section.pending().unwrap();
788	assert_eq!(
789	pending[1..]
790	.iter()
791	.map(|edit| (edit.id, &edit.edit))
792	.collect::<Vec<_>>(),
793	[(ids[0], &edits[0]), (ids[1], &edits[1])]
794	);
795
796	// An edit that could never be written is refused alone; the next one lands.
797	replica
798	.lock()
799	.unwrap()
800	.execute_batch(
801	"CREATE TEMP TRIGGER refuse BEFORE INSERT ON main.edits WHEN NEW.author = 'Refused'
802	BEGIN SELECT RAISE(ABORT, 'Disposable refusal'); END;",
803	)
804	.unwrap();
805	for (author, colour) in [("Refused", 0x0012_3456), ("Fixture", 0x0065_4321)] {
806	let replied = replied.clone();
807	section
808	.apply_with(
809	author,
810	Edit {
811	at: crate::now(),
812	ops: vec![Op::Page {
813	space,
814	op: onestore::op::PageOp::Color(Some(colour)),
815	}],
816	},
817	move |result| {
818	replied
819	.send(result.map_err(|error| error.to_string()))
820	.unwrap();
821	},
822	)
823	.unwrap();
824	}
825	assert!(
826	replies
827	.recv_timeout(Duration::from_secs(10))
828	.unwrap()
829	.is_err()
830	);
831	let last = replies
832	.recv_timeout(Duration::from_secs(10))
833	.unwrap()
834	.unwrap();
835	section.written().unwrap();
836	assert_eq!(section.pending().unwrap().last().unwrap().id, last);
837	assert_eq!(section.page(space).unwrap().color, Some(0x0065_4321));
838	section.wake();
839	let deadline = Instant::now() + Duration::from_secs(10);
840	while !section.events().iter().any(|event| {
841	matches!(event, Event::Attempt { id, status: EditStatus::Published { .. } } if *id == last)
842	}) {
843	assert!(Instant::now() < deadline);
844	std::thread::sleep(Duration::from_millis(10));
845	}
846	section.close().unwrap();
847	drop(replica);
848	let pages = stored_pages(&fs::read(&file).unwrap()).unwrap();
849	let waited: Vec<_> = pages
850	.iter()
851	.filter(|page| page.page.title == title)
852	.collect();
853	assert_eq!(waited.len(), 1);
854	assert_eq!(waited[0].page.color, Some(0x0065_4321));
855	assert!(Replica::open(&cache).unwrap().pending().unwrap().is_empty());
856	}