clover's git

snowbound

1	//! The sections of an open notebook that no session holds, kept in sync as OneNote 2010
2	//! keeps every section of an open notebook: queued edits publish and other clients' changes
3	//! are noticed without the section being open.
4
5	use crate::fs;
6	use crate::{
7	EditStatus, Error, Remote, Replica, Result,
8	discover::{Entry, Listed},
9	session::{Observed, Section, SyncStatus, reached, sync_status},
10	worker::Signal,
11	};
12	use onestore::Stamp;
13	use std::{
14	collections::{BTreeMap, BTreeSet},
15	io,
16	path::{Path, PathBuf},
17	sync::{
18	Arc, Mutex, Weak,
19	atomic::{AtomicBool, Ordering},
20	},
21	time::Duration,
22	};
23	use web_time::Instant;
24
25	/// How long a section that could not be reached waits to be tried again; OneNote 2010
26	/// retries a failing section about every 31 seconds.
27	const RETRY: Duration = Duration::from_secs(31);
28	/// Between one section's first check and the next's, so opening a notebook reads its files
29	/// one after another rather than in a burst.
30	const STAGGER: Duration = Duration::from_millis(100);
31	/// How long a reported change settles before its section is checked, so the writes of one
32	/// commit cost one check.
33	const SETTLE: Duration = Duration::from_secs(1);
34
35	/// Keeps each watched section in sync: a section is checked when a watch on the notebook's
36	/// folder reports it changed, when it could not be reached, and otherwise every interval.
37	/// On connecting, and when a watch reports a folder rather than a file, the folders are
38	/// listed and only the sections listed otherwise than before are checked, as OneNote 2010
39	/// reopens a notebook. A check costs one stamp read; a section whose replica has edits waiting,
40	/// or whose file moved past the replica's base, has its replica opened for the synchronization
41	/// steps that publish or rebase it, then closed again. A section without a replica gets one,
42	/// its offline copy, where the notebook keeps them. A section a session holds is left to that
43	/// session's worker, which the watch wakes. Dropping requests cancellation without waiting for
44	/// the step in flight.
45	pub struct Background(Arc<Shared>, Mutex<Option<crate::task::JoinHandle<()>>>);
46
47	/// A section for `Background::watch`, as its notebook knows it.
48	pub struct Known {
49	/// The catalog path.
50	pub path: String,
51	pub replica: Option<PathBuf>,
52	/// How the notebook's discovery last found the file: its listing, and its stamp then.
53	pub found: Option<(Listed, Stamp)>,
54	/// The file as that discovery read it, if it did, which the first check takes in place
55	/// of reading the file again while its stamp still matches.
56	pub image: Option<Vec<u8>>,
57	}
58
59	/// What the background thread and whatever reports changes to it share.
60	pub(crate) struct Shared {
61	signal: Arc<Signal>,
62	watched: Mutex<Watched>,
63	/// Asked for by `discard`, once the thread stops.
64	discard: AtomicBool,
65	/// Hears every report of changed paths (`on_touched`).
66	listener: Mutex<Option<Listener>>,
67	}
68
69	/// Hears the catalog paths a report names.
70	pub type Listener = Box<dyn Fn(&[String]) + Send + Sync>;
71
72	/// A connection's report of its notebook folder's changes, from a watch armed on connecting.
73	#[cfg_attr(not(any(feature = "smb", feature = "live")), allow(dead_code))]
74	pub(crate) struct Reports {
75	shared: Weak<Shared>,
76	connection: u64,
77	}
78
79	#[derive(Default)]
80	struct Watched {
81	sections: Vec<Watch>,
82	/// Sections or cached folders changed since `changed` was last asked.
83	changed: Vec<String>,
84	/// Folders a watch reported without naming the file, listed at `relist`.
85	folders: BTreeSet<String>,
86	relist: Option<Instant>,
87	/// Counts connections, so a watch that ended with an earlier one is not this one's.
88	connection: u64,
89	/// The current connection's watch ended, or none was connected: changes since went
90	/// unreported.
91	lost: bool,
92	/// The current connection's watch reports the folder's changes.
93	reported: bool,
94	/// How long a report settles, where not `SETTLE` (`Background::set_settle`).
95	settle: Option<Duration>,
96	}
97
98	struct Watch {
99	path: String,
100	replica: Option<PathBuf>,
101	/// The file's stamp when last read.
102	stamp: Option<Stamp>,
103	/// How the file was listed when its folder was last listed; the stamp was read since.
104	listed: Option<Listed>,
105	/// The folder's listing since `stamp` shows the file unchanged, so the next check need not
106	/// read it.
107	current: bool,
108	/// A watch reported the file changed since its folder's listing began.
109	reported: bool,
110	status: SyncStatus,
111	/// The file as discovery read it, until the next check takes it.
112	image: Option<Vec<u8>>,
113	/// When the section is next checked.
114	due: Instant,
115	held: Option<Held>,
116	}
117
118	#[derive(Clone)]
119	struct Held {
120	worker: Weak<Signal>,
121	replica: Weak<Replica>,
122	observed: Weak<Mutex<Observed>>,
123	}
124
125	impl Background {
126	/// How often each section is checked while a watch reports its folder's changes, against
127	/// a report that went missing; OneNote 2010, watching, checks nothing on a timer.
128	pub const BACKSTOP: Duration = Duration::from_secs(60 * 60);
129	/// How often each section is checked where nothing reports its folder's changes.
130	pub const UNWATCHED: Duration = Duration::from_secs(15);
131
132	/// Starts the thread. `connect` answers how the connection reaches each section file and
133	/// lists each folder, both by catalog path, arming any watch that reports the folder's
134	/// changes through `Reports`, and whether one does; it runs again after a transport
135	/// failure. `copies` keeps an offline copy of every section.
136	pub(crate) fn start<R, B, L, LF>(
137	copies: bool,
138	mut connect: impl FnMut(Reports) -> io::Result<((B, L), bool)> + Send + 'static,
139	notify: impl Fn() + Send + 'static,
140	) -> Result<Self>
141	where
142	R: Remote,
143	B: FnMut(&str) -> R,
144	L: FnMut(String) -> LF,
145	LF: Future<Output = io::Result<Vec<Entry>>>,
146	{
147	let (signal, receiver) = Signal::new();
148	let shared = Arc::new(Shared {
149	signal,
150	watched: Mutex::default(),
151	discard: AtomicBool::new(false),
152	listener: Mutex::default(),
153	});
154	let weak = Arc::downgrade(&shared);
155	let owner = Arc::clone(&shared);
156	let thread = crate::task::spawn("onestore-background", move || async move {
157	let signal = &owner.signal;
158	let mut bound: Option<(B, L)> = None;
159	let mut news = false;
160	while !signal.stopped.load(Ordering::Acquire) {
161	if signal.offline.load(Ordering::Acquire)
162	&& !signal.requested.load(Ordering::Acquire)
163	{
164	// Working online again lists every folder again.
165	if let Ok(mut watched) = owner.watched.lock() {
166	watched.watching(false);
167	watched.lost = true;
168	}
169	bound = None;
170	crate::task::wait(&receiver, None).await;
171	continue;
172	}
173	let now = Instant::now();
174	let next = {
175	let Ok(mut watched) = owner.watched.lock() else {
176	return;
177	};
178	if std::mem::take(&mut watched.lost) {
179	bound = None;
180	watched.watching(false);
181	for watch in &mut watched.sections {
182	watch.due = now;
183	}
184	}
185	watched.next(now, bound.is_some())
186	};
187	let (path, replica, seen, current, image) = match next {
188	Next::Wait(wait) => {
189	signal.requested.store(false, Ordering::Release);
190	if std::mem::take(&mut news) {
191	notify();
192	}
193	crate::task::wait(&receiver, wait).await;
194	continue;
195	}
196	Next::Connect(connection) => {
197	let reports = Reports {
198	shared: weak.clone(),
199	connection: connection + 1,
200	};
201	let connected = connect(reports);
202	let folders = {
203	let Ok(mut watched) = owner.watched.lock() else {
204	return;
205	};
206	watched.connection += 1;
207	watched.lost = false;
208	match &connected {
209	Ok((_, reported)) => {
210	watched.watching(*reported);
211	Some(watched.rescanning())
212	}
213	Err(error) => {
214	let error = Error::RemoteIo(io::Error::new(
215	error.kind(),
216	error.to_string(),
217	));
218	let retry = now + RETRY;
219	for watch in &mut watched.sections {
220	news |= watch.fail(&error, None);
221	watch.due = watch.due.max(retry);
222	}
223	watched.relist = watched.relist.map(|due| due.max(retry));
224	None
225	}
226	}
227	};
228	if let (Ok((mut files, _)), Some(folders)) = (connected, folders) {
229	let listings = list(&mut files.1, folders).await;
230	let Ok(mut watched) = owner.watched.lock() else {
231	return;
232	};
233	watched.listed(&listings, Some(copies), Instant::now());
234	bound = Some(files);
235	}
236	continue;
237	}
238	Next::List(folders) => {
239	let Some((_, list_folder)) = &mut bound else {
240	continue;
241	};
242	let listings = list(list_folder, folders).await;
243	let Ok(mut watched) = owner.watched.lock() else {
244	return;
245	};
246	watched.listed(&listings, None, Instant::now());
247	continue;
248	}
249	Next::Check {
250	path,
251	replica,
252	seen,
253	current,
254	image,
255	} => (path, replica, seen, current, image),
256	};
257	let Some((bind, _)) = &mut bound else {
258	continue;
259	};
260	let (queued, outcome) = step_async(
261	&mut bind(&path),
262	replica.as_deref(),
263	seen.as_deref(),
264	current,
265	image,
266	copies,
267	)
268	.await;
269	if outcome.as_ref().is_err_and(disconnected) {
270	bound = None;
271	}
272	let Ok(mut watched) = owner.watched.lock() else {
273	return;
274	};
275	let interval = watched.interval();
276	let Watched {
277	sections, changed, ..
278	} = &mut *watched;
279	let Some(watch) = sections.iter_mut().find(|watch| watch.path == path) else {
280	continue;
281	};
282	let now = Instant::now();
283	watch.current = false;
284	news |= match outcome {
285	Ok((stamp, moved)) => {
286	watch.stamp = Some(stamp);
287	watch.due = watch.due.min(now + interval);
288	if moved {
289	changed.push(path);
290	}
291	let before = summary(&watch.status);
292	if let Some(queued) = queued {
293	watch.status = SyncStatus {
294	synced: Some(crate::now()),
295	error: None,
296	queued,
297	};
298	}
299	moved || before != summary(&watch.status)
300	}
301	Err(error) => {
302	watch.due = watch.due.min(now + RETRY);
303	watch.fail(&error, queued)
304	}
305	};
306	}
307	if let Ok(mut watched) = owner.watched.lock() {
308	watched.watching(false);
309	if owner.discard.load(Ordering::Acquire) {
310	for replica in watched
311	.sections
312	.iter()
313	.filter_map(|watch| watch.replica.as_ref())
314	{
315	discard(replica);
316	}
317	}
318	}
319	})?;
320	Ok(Self(shared, Mutex::new(Some(thread))))
321	}
322
323	/// Keeps the sections of a notebook on a share in sync while they are not open, with an
324	/// offline copy of each, and one watch on the notebook's folder reporting what changed,
325	/// as OneNote 2010 watches it; `connect` runs again after a transport failure. Paths are
326	/// relative to `root`.
327	#[cfg(feature = "smb")]
328	pub fn smb(
329	root: &str,
330	limit: usize,
331	mut connect: impl FnMut() -> io::Result<crate::smb::Client> + Send + 'static,
332	notify: impl Fn() + Send + 'static,
333	) -> Result<Self> {
334	let root = root.replace('\\', "/");
335	Self::start(
336	true,
337	move |reports| {
338	let client = Arc::new(connect()?);
339	let reported = match client.watch(&root, move |changed| match changed {
340	Ok(paths) => reports.touched(&paths),
341	Err(_) => reports.lost(),
342	}) {
343	Ok(()) => true,
344	Err(error) if error.kind() == io::ErrorKind::Unsupported => false,
345	Err(error) => return Err(error),
346	};
347	let (bound, folder) = (Arc::clone(&client), root.clone());
348	let bind = move |path: &str| {
349	let file = match folder.as_str() {
350	"" => path.to_owned(),
351	root => format!("{root}/{path}"),
352	};
353	crate::SmbRemote::new(Arc::clone(&bound), file, limit)
354	};
355	let root = root.clone();
356	let list = move |folder: String| {
357	std::future::ready((|| {
358	use crate::discover::Source;
359	crate::discover::Smb::new(&client, &root)?
360	.entries(&folder, crate::session::LIMITS.entries)
361	})())
362	};
363	Ok(((bind, list), reported))
364	},
365	notify,
366	)
367	}
368
369	/// Keeps the sections of a notebook a Live Share host serves in sync while they are not
370	/// open, with an offline copy of each; the host reports what changed. Paths are catalog
371	/// paths.
372	#[cfg(feature = "live")]
373	pub fn hosted(
374	guest: Arc<crate::live::share::Guest>,
375	notify: impl Fn() + Send + 'static,
376	) -> Result<Self> {
377	let background = Self::start(
378	true,
379	move |reports| {
380	guest.watch(reports)?;
381	let (bound, listing) = (Arc::clone(&guest), Arc::clone(&guest));
382	let bind = move |path: &str| crate::live::share::HostedRemote::new(&bound, path);
383	let list = move |folder: String| {
384	let listing = listing.clone();
385	async move { listing.entries_async(&folder).await }
386	};
387	Ok(((bind, list), true))
388	},
389	notify,
390	)?;
391	background.set_settle(crate::live::share::SETTLE);
392	Ok(background)
393	}
394
395	/// Keeps the sections of a notebook a document provider keeps in sync while they are not
396	/// open, with an offline copy of each. Nothing reports the folder's changes; the host
397	/// passes what it hears of them to `touched`. Paths are catalog paths.
398	pub fn provider(
399	documents: Arc<crate::provider::Documents>,
400	notify: impl Fn() + Send + 'static,
401	) -> Result<Self> {
402	Self::start(
403	true,
404	move |_| {
405	let (bound, listing) = (Arc::clone(&documents), Arc::clone(&documents));
406	let bind = move |path: &str| crate::provider::ProviderRemote::new(&bound, path);
407	let list = move |folder: String| std::future::ready(listing.entries(&folder));
408	Ok(((bind, list), false))
409	},
410	notify,
411	)
412	}
413
414	/// Checks what a report names after `settle` rather than a second, where reports come
415	/// one to a commit, as a Live Share host sends them.
416	pub fn set_settle(&self, settle: Duration) {
417	if let Ok(mut watched) = self.0.watched.lock() {
418	watched.settle = Some(settle);
419	}
420	}
421
422	/// Has `listener` hear every report of changed paths from now on, as a Live Share host
423	/// passes them to its guests; `None` stops.
424	pub fn on_touched(&self, listener: Option<Listener>) {
425	if let Ok(mut held) = self.0.listener.lock() {
426	*held = listener;
427	}
428	}
429
430	/// Watches these sections (`Notebook::replicas`); a section not watched before is first
431	/// checked soon after, the next one a little later.
432	pub fn watch(&self, sections: Vec<Known>) {
433	if let Ok(mut watched) = self.0.watched.lock() {
434	watched.watch(sections, Instant::now());
435	}
436	self.0.signal.wake();
437	}
438
439	/// Leaves the section at catalog `path` to `section`'s worker, which from now on the watch
440	/// on the notebook's folder wakes when the file changes, instead of the worker checking
441	/// it every few seconds. Once the section closes, this takes it over again.
442	pub fn hold(&self, path: &str, section: &Section) {
443	let replica = section.replica();
444	let Some(worker) = replica
445	.section
446	.worker
447	.lock()
448	.ok()
449	.and_then(|worker| worker.upgrade())
450	else {
451	return;
452	};
453	if let Ok(mut watched) = self.0.watched.lock() {
454	let reported = watched.reported;
455	if let Some(watch) = watched.sections.iter_mut().find(|watch| watch.path == path) {
456	worker.watched.store(reported, Ordering::Release);
457	watch.held = Some(Held {
458	worker: Arc::downgrade(&worker),
459	replica: Arc::downgrade(replica),
460	observed: Arc::downgrade(&section.observed),
461	});
462	}
463	}
464	let (shared, path) = (Arc::downgrade(&self.0), path.to_owned());
465	if let Ok(mut released) = replica.section.released.0.lock() {
466	*released = Some(Box::new(move || {
467	if let Some(shared) = shared.upgrade() {
468	shared.released(&path);
469	}
470	}));
471	}
472	}
473
474	/// Checks the sections at or below these catalog paths soon, as a watch on the notebook's
475	/// folder reports them changed: a section's own path checks it, a folder's lists it first
476	/// and checks the sections listed otherwise. `""` names the notebook's folder.
477	pub fn touched(&self, paths: &[String]) {
478	self.0.touched(paths);
479	}
480
481	/// Each watched section's status, in watch order: a held section's live state while its
482	/// session syncs, otherwise this thread's last check of it.
483	pub fn status(&self) -> Vec<(String, SyncStatus)> {
484	let sections: Vec<_> = self.0.watched.lock().map_or_else(
485	|_| Vec::new(),
486	|watched| {
487	watched
488	.sections
489	.iter()
490	.map(|watch| {
491	let status = &watch.status;
492	(
493	watch.path.clone(),
494	SyncStatus {
495	synced: status.synced,
496	error: status
497	.error
498	.as_ref()
499	.map(|error| io::Error::new(error.kind(), error.to_string())),
500	queued: status.queued,
501	},
502	watch.held.clone(),
503	)
504	})
505	.collect()
506	},
507	);
508	sections
509	.into_iter()
510	.map(|(path, fallback, held)| {
511	let live = held.and_then(|held| {
512	let worker = (held.worker.upgrade())
513	.filter(|worker| !worker.stopped.load(Ordering::Acquire))?;
514	let replica = held.replica.upgrade()?;
515	let observed = held.observed.upgrade()?;
516	let synced =
517	|| Some(worker.synced.load(Ordering::Acquire)).filter(|time| *time != 0);
518	Some(sync_status(&replica, synced, &observed))
519	});
520	let status = match live {
521	Some(Ok(status)) => status,
522	Some(Err(error)) => SyncStatus {
523	error: Some(reached(&error)),
524	..fallback
525	},
526	None => fallback,
527	};
528	(path, status)
529	})
530	.collect()
531	}
532
533	/// Sections whose file changed since the last call, by catalog path.
534	pub fn changed(&self) -> Vec<String> {
535	self.0
536	.watched
537	.lock()
538	.map(|mut watched| std::mem::take(&mut watched.changed))
539	.unwrap_or_default()
540	}
541
542	/// Reads every section's stamp now, working offline included (Sync Now).
543	pub fn wake(&self) {
544	if let Ok(mut watched) = self.0.watched.lock() {
545	let now = Instant::now();
546	for watch in &mut watched.sections {
547	watch.due = now;
548	watch.current = false;
549	}
550	}
551	self.0.signal.requested.store(true, Ordering::Release);
552	self.0.signal.wake();
553	}
554
555	/// Stops for good, then deletes each section's replica that holds nothing unpublished, as
556	/// OneNote lets go of a notebook it closes; one with edits waiting, or that a session
557	/// holds, stays.
558	pub fn discard(&self) {
559	self.0.discard.store(true, Ordering::Release);
560	self.0.signal.stopped.store(true, Ordering::Release);
561	self.0.signal.wake();
562	}
563
564	/// Stops future checks; native threads finish the current step before returning.
565	pub fn stop(&self) {
566	self.0.signal.stopped.store(true, Ordering::Release);
567	self.0.signal.wake();
568	let thread = self.1.lock().ok().and_then(|mut thread| thread.take());
569	#[cfg(not(target_arch = "wasm32"))]
570	if let Some(thread) = thread {
571	let _ = thread.join();
572	}
573	#[cfg(target_arch = "wasm32")]
574	drop(thread);
575	}
576
577	/// Stops future checks and awaits the current step before files can move.
578	pub async fn stop_async(&self) {
579	#[cfg(not(target_arch = "wasm32"))]
580	self.stop();
581	#[cfg(target_arch = "wasm32")]
582	{
583	self.0.signal.stopped.store(true, Ordering::Release);
584	self.0.signal.wake();
585	let thread = self.1.lock().ok().and_then(|mut thread| thread.take());
586	if let Some(thread) = thread {
587	let _ = thread.join().await;
588	}
589	}
590	}
591
592	/// Working offline, nothing is checked until `wake`, or until working online again, which
593	/// lists every folder again.
594	pub fn set_offline(&self, offline: bool) {
595	self.0.signal.offline.store(offline, Ordering::Release);
596	self.0.signal.wake();
597	}
598	}
599
600	impl Drop for Background {
601	fn drop(&mut self) {
602	self.0.signal.stopped.store(true, Ordering::Release);
603	self.0.signal.wake();
604	}
605	}
606
607	impl Shared {
608	fn touched(&self, paths: &[String]) {
609	if let Ok(mut watched) = self.watched.lock() {
610	watched.touched(paths, Instant::now());
611	}
612	self.signal.wake();
613	if let Ok(listener) = self.listener.lock()
614	&& let Some(listener) = &*listener
615	{
616	listener(paths);
617	}
618	}
619
620	/// The session holding the section at `path` let it go: it is checked now.
621	fn released(&self, path: &str) {
622	if let Ok(mut watched) = self.watched.lock()
623	&& let Some(watch) = watched.sections.iter_mut().find(|watch| watch.path == path)
624	{
625	watch.held = None;
626	watch.current = false;
627	watch.due = Instant::now();
628	}
629	self.signal.wake();
630	}
631	}
632
633	#[cfg_attr(not(any(feature = "smb", feature = "live")), allow(dead_code))]
634	impl Reports {
635	#[cfg(all(feature = "live", target_arch = "wasm32"))]
636	pub(crate) fn catalog(&self, folder: &str) {
637	if let Some(shared) = self.shared.upgrade()
638	&& let Ok(mut watched) = shared.watched.lock()
639	&& !watched.changed.iter().any(|path| path == folder)
640	{
641	watched.changed.push(folder.to_owned());
642	}
643	}
644
645	/// The sections at or below these paths changed.
646	pub(crate) fn touched(&self, paths: &[String]) {
647	if let Some(shared) = self.shared.upgrade() {
648	shared.touched(paths);
649	}
650	}
651
652	/// The watch ended, and with it, as far as anyone can tell, the connection.
653	pub(crate) fn lost(&self) {
654	if let Some(shared) = self.shared.upgrade() {
655	if let Ok(mut watched) = shared.watched.lock()
656	&& watched.connection == self.connection
657	{
658	watched.lost = true;
659	}
660	shared.signal.wake();
661	}
662	}
663	}
664
665	/// What the background thread does next.
666	enum Next {
667	Wait(Option<Duration>),
668	/// Connects, as the current connection is `.0`, then lists every folder.
669	Connect(u64),
670	/// Lists these folders, checking the sections listed otherwise.
671	List(BTreeSet<String>),
672	Check {
673	path: String,
674	replica: Option<PathBuf>,
675	seen: Option<Box<Stamp>>,
676	current: bool,
677	image: Option<Vec<u8>>,
678	},
679	}
680
681	impl Watched {
682	fn watch(&mut self, sections: Vec<Known>, now: Instant) {
683	let mut previous = std::mem::take(&mut self.sections);
684	let mut new = 0;
685	self.sections = sections
686	.into_iter()
687	.map(
688	|known| match previous.iter().position(|watch| watch.path == known.path) {
689	Some(index) => Watch {
690	replica: known.replica,
691	image: known.image,
692	..previous.swap_remove(index)
693	},
694	None => {
695	new += 1;
696	let (listed, stamp) = known.found.unzip();
697	Watch {
698	path: known.path,
699	replica: known.replica,
700	stamp,
701	listed,
702	current: false,
703	reported: false,
704	status: SyncStatus {
705	synced: None,
706	error: None,
707	queued: 0,
708	},
709	image: known.image,
710	due: now + STAGGER * (new - 1),
711	held: None,
712	}
713	}
714	},
715	)
716	.collect();
717	}
718
719	fn interval(&self) -> Duration {
720	if self.reported {
721	Background::BACKSTOP
722	} else {
723	Background::UNWATCHED
724	}
725	}
726
727	/// Whether a watch reports the folder's changes from now on, as the held sessions' workers
728	/// learn.
729	fn watching(&mut self, reported: bool) {
730	self.reported = reported;
731	for signal in self
732	.sections
733	.iter()
734	.filter_map(|watch| watch.held.as_ref()?.worker.upgrade())
735	{
736	signal.watched.store(reported, Ordering::Release);
737	signal.wake();
738	}
739	}
740
741	fn touched(&mut self, paths: &[String], now: Instant) {
742	let settled = now + self.settle.unwrap_or(SETTLE);
743	for path in paths {
744	let path = path.trim_matches('/');
745	let section = self
746	.sections
747	.iter()
748	.position(|watch| watch.path.eq_ignore_ascii_case(path));
749	match section {
750	Some(index) => {
751	let watch = &mut self.sections[index];
752	watch.due = watch.due.min(settled);
753	watch.current = false;
754	watch.reported = true;
755	}
756	None if self.sections.iter().any(|watch| within(&watch.path, path)) => {
757	self.folders.insert(path.to_owned());
758	self.relist = Some(self.relist.map_or(settled, |due| due.min(settled)));
759	}
760	None => {}
761	}
762	}
763	}
764
765	/// Every folder holding a section, to list them all, as a listing begins.
766	fn rescanning(&mut self) -> BTreeSet<String> {
767	for watch in &mut self.sections {
768	watch.reported = false;
769	}
770	self.parents(&[String::new()])
771	}
772
773	/// The folders holding the sections at or below `paths`.
774	fn parents<'a>(&self, paths: impl IntoIterator<Item = &'a String> + Clone) -> BTreeSet<String> {
775	self.sections
776	.iter()
777	.filter(|watch| {
778	paths
779	.clone()
780	.into_iter()
781	.any(|path| within(&watch.path, path))
782	})
783	.map(|watch| split(&watch.path).0.to_owned())
784	.collect()
785	}
786
787	/// Takes the folders' listings. With `rescan`, as on connecting (whether copies are kept),
788	/// every section of a listed folder is checked, from now: first those whose file lists as
789	/// when its stamp was last read, which need no reading unless for a copy, then the others
790	/// one `STAGGER` after another. Otherwise only the sections listed otherwise are, as are
791	/// those a watch reported meanwhile either way.
792	fn listed(
793	&mut self,
794	listings: &BTreeMap<String, Vec<Entry>>,
795	rescan: Option<bool>,
796	now: Instant,
797	) {
798	let mut staggered = 0;
799	for watch in &mut self.sections {
800	let (folder, name) = split(&watch.path);
801	let Some(entries) = listings.get(folder) else {
802	continue;
803	};
804	let listed = entries
805	.iter()
806	.find(|entry| entry.name.eq_ignore_ascii_case(name))
807	.map(|entry| entry.listed);
808	let unchanged = watch.stamp.is_some() && listed.is_some() && listed == watch.listed;
809	watch.listed = listed;
810	match rescan {
811	// A report while listing may name a change the listing missed.
812	Some(_) if watch.reported => watch.current = false,
813	Some(copies) => {
814	watch.current = unchanged;
815	let read = !unchanged
816	|| copies
817	&& watch.image.is_none()
818	&& watch
819	.replica
820	.as_ref()
821	.is_some_and(|replica| fs::metadata(replica).is_err());
822	watch.due = if read {
823	staggered += 1;
824	now + STAGGER * (staggered - 1)
825	} else {
826	now
827	};
828	}
829	None if !unchanged => {
830	watch.current = false;
831	watch.due = watch.due.min(now);
832	}
833	None => {}
834	}
835	}
836	}
837
838	/// What to do at `now`, waking the held sections that are due; `bound` while connected.
839	fn next(&mut self, now: Instant, bound: bool) -> Next {
840	let interval = self.interval();
841	loop {
842	let due = (0..self.sections.len()).min_by_key(|&index| self.sections[index].due);
843	let at = due.map(|index| self.sections[index].due);
844	if let Some(relist) = self.relist
845	&& relist <= now
846	&& at.is_none_or(|at| relist <= at)
847	{
848	if !bound {
849	return Next::Connect(self.connection);
850	}
851	let folders = std::mem::take(&mut self.folders);
852	self.relist = None;
853	return Next::List(self.parents(&folders));
854	}
855	let Some(index) = due.filter(|&index| self.sections[index].due <= now) else {
856	let wait = [at, self.relist].into_iter().flatten().min();
857	return Next::Wait(wait.map(|at| at.saturating_duration_since(now)));
858	};
859	let watch = &mut self.sections[index];
860	if !bound {
861	return Next::Connect(self.connection);
862	}
863	if let Some(worker) = watch.held.as_ref().and_then(|held| held.worker.upgrade()) {
864	worker.wake();
865	watch.due = now + interval;
866	continue;
867	}
868	// Anything asking for the section during the check can only bring this forward.
869	watch.due = now + interval.max(RETRY);
870	return Next::Check {
871	path: watch.path.clone(),
872	replica: watch.replica.clone(),
873	seen: watch.stamp.clone().map(Box::new),
874	current: watch.current,
875	image: watch.image.take(),
876	};
877	}
878	}
879	}
880
881	/// Lists each of `folders` through `list`; a folder that cannot be listed lists nothing, so
882	/// that its sections are checked.
883	async fn list<F: Future<Output = io::Result<Vec<Entry>>>>(
884	list: &mut impl FnMut(String) -> F,
885	folders: BTreeSet<String>,
886	) -> BTreeMap<String, Vec<Entry>> {
887	let mut listings = BTreeMap::new();
888	for folder in folders {
889	let entries = list(folder.clone()).await.unwrap_or_default();
890	listings.insert(folder, entries);
891	}
892	listings
893	}
894
895	/// Deletes the replica at `replica` if it holds nothing unpublished and no one holds it.
896	fn discard(replica: &Path) {
897	if matches!(
898	crate::closed(replica).and_then(|held| crate::peek(&held)),
899	Ok((_, 0))
900	) {
901	for suffix in ["-wal", "-shm", ""] {
902	let mut file = replica.as_os_str().to_owned();
903	file.push(suffix);
904	let _ = fs::remove_file(file);
905	}
906	}
907	}
908
909	/// Whether the catalog path `section` is `path` or lies below it, as a share compares names.
910	fn within(section: &str, path: &str) -> bool {
911	let path = path.trim_matches('/');
912	path.is_empty()
913	|| section.len() >= path.len()
914	&& section.is_char_boundary(path.len())
915	&& section[..path.len()].eq_ignore_ascii_case(path)
916	&& matches!(section.as_bytes().get(path.len()), None | Some(b'/'))
917	}
918
919	/// A catalog path's folder and name.
920	fn split(path: &str) -> (&str, &str) {
921	path.rsplit_once('/').unwrap_or(("", path))
922	}
923
924	/// Whether a failed step lost the connection, rather than failing for its one file.
925	fn disconnected(error: &Error) -> bool {
926	use io::ErrorKind::*;
927	let kind = match error {
928	Error::RemoteIo(error) => error.kind(),
929	Error::Remote(error) => error.error.kind(),
930	_ => return false,
931	};
932	matches!(
933	kind,
934	NotConnected | TimedOut | ConnectionReset | ConnectionAborted | BrokenPipe
935	)
936	}
937
938	impl Watch {
939	/// Records a failed step and what it found waiting, answering whether the status shown
940	/// changes.
941	fn fail(&mut self, error: &Error, queued: Option<u64>) -> bool {
942	let before = summary(&self.status);
943	self.status.error = Some(reached(error));
944	self.status.queued = queued.unwrap_or(self.status.queued);
945	before != summary(&self.status)
946	}
947	}
948
949	/// What the host shows of a status.
950	fn summary(status: &SyncStatus) -> (bool, Option<io::ErrorKind>, u64) {
951	(
952	status.synced.is_some(),
953	status.error.as_ref().map(io::Error::kind),
954	status.queued,
955	)
956	}
957
958	/// One section's step: how many of its edits wait (`None` while unknown, as while a session
959	/// holds its replica), then the file's stamp now and whether it changed since `seen`, which
960	/// with `current` is the stamp now, unread. With `copies`, a section without a replica gets
961	/// one from the file as it is now. `image`, the file as discovery read it, stands in for
962	/// reading it while the stamp is still its own.
963	async fn step_async<R: Remote>(
964	remote: &mut R,
965	replica: Option<&Path>,
966	seen: Option<&Stamp>,
967	current: bool,
968	image: Option<Vec<u8>>,
969	copies: bool,
970	) -> (Option<u64>, Result<(Stamp, bool)>) {
971	let stamp = match seen.filter(|_| current) {
972	Some(seen) => seen.clone(),
973	None => match crate::sync::awaited(remote, Remote::stamp).await {
974	Ok(stamp) => stamp,
975	Err(error) => return (None, Err(Error::RemoteIo(error))),
976	},
977	};
978	let remote = &mut Discovered {
979	image: image.and_then(|image| Some((Stamp::of(&image).ok()?, image))),
980	stamp: stamp.clone(),
981	remote,
982	};
983	let moved = seen.is_some_and(|seen| *seen != stamp);
984	let Some(replica) = replica else {
985	return (Some(0), Ok((stamp, moved)));
986	};
987	if fs::metadata(replica).is_err() {
988	if !copies {
989	return (Some(0), Ok((stamp, moved)));
990	}
991	let copied = (async {
992	let image = crate::sync::awaited(remote, Remote::read)
993	.await
994	.map_err(Error::RemoteIo)?;
995	if let Some(folder) = replica.parent() {
996	fs::create_dir_all(folder)?;
997	}
998	Replica::seed(replica, &image, None)?;
999	fs::durable().await?;
1000	Ok(Stamp::of(&image)?)
1001	})
1002	.await;
1003	return match copied {
1004	Ok(stamp) => (Some(0), Ok((stamp, moved))),
1005	// A session made it first.
1006	Err(Error::Io(error)) if error.kind() == io::ErrorKind::AlreadyExists => {
1007	(None, Ok((stamp, false)))
1008	}
1009	Err(error) => (None, Err(error)),
1010	};
1011	}
1012	// A version kept beside the file leaves its stamp as it was.
1013	let versions = match crate::sync::awaited(remote, Remote::versions).await {
1014	Ok(versions) => versions,
1015	Err(error) => return (None, Err(Error::RemoteIo(error))),
1016	};
1017	match crate::closed(replica).and_then(|held| crate::peek(&held)) {
1018	Ok((base, 0)) if base == stamp && versions.is_empty() => {
1019	return (Some(0), Ok((stamp, moved)));
1020	}
1021	Err(error) if error.busy() => return (None, Ok((stamp, false))),
1022	_ => {}
1023	}
1024	let replica = match Replica::open(replica) {
1025	Ok(replica) => replica,
1026	Err(error) if error.busy() => return (None, Ok((stamp, false))),
1027	Err(error) => return (None, Err(error)),
1028	};
1029	let synced = (async {
1030	let mut changed = moved;
1031	loop {
1032	let synced = replica.sync_once_async(remote).await?;
1033	changed |= !synced.changed.is_empty();
1034	if !matches!(synced.edit, Some((_, EditStatus::Published { .. }))) {
1035	return Ok((
1036	crate::sync::awaited(remote, Remote::stamp)
1037	.await
1038	.map_err(Error::RemoteIo)?,
1039	changed,
1040	));
1041	}
1042	}
1043	})
1044	.await;
1045	let queued = replica
1046	.recovery_summary()
1047	.ok()
1048	.map(|summary| summary.queued_edits);
1049	(queued, synced)
1050	}
1051
1052	/// A remote whose first read, while the file's stamp is still `image`'s, answers `image`.
1053	struct Discovered<'a, R> {
1054	remote: &'a mut R,
1055	image: Option<(Stamp, Vec<u8>)>,
1056	/// The stamp last read.
1057	stamp: Stamp,
1058	}
1059
1060	impl<R: Remote> Remote for Discovered<'_, R> {
1061	fn pending(&mut self) -> Option<std::pin::Pin<Box<dyn Future<Output = ()> + '_>>> {
1062	self.remote.pending()
1063	}
1064
1065	fn read(&mut self) -> io::Result<Vec<u8>> {
1066	match self.image.take() {
1067	Some((stamp, image)) if stamp == self.stamp => Ok(image),
1068	_ => self.remote.read(),
1069	}
1070	}
1071
1072	fn stamp(&mut self) -> io::Result<Stamp> {
1073	self.stamp = self.remote.stamp()?;
1074	Ok(self.stamp.clone())
1075	}
1076
1077	fn publish(
1078	&mut self,
1079	transaction: &onestore::Transaction,
1080	) -> std::result::Result<(), onestore::CommitError> {
1081	self.remote.publish(transaction)
1082	}
1083
1084	fn confirm(&mut self, base: &Stamp) -> std::result::Result<(), onestore::CommitError> {
1085	self.remote.confirm(base)
1086	}
1087
1088	fn versions(&mut self) -> io::Result<Vec<crate::Version>> {
1089	self.remote.versions()
1090	}
1091
1092	fn version(&mut self, id: &str) -> io::Result<Vec<u8>> {
1093	self.remote.version(id)
1094	}
1095
1096	fn retire(&mut self, id: &str, keep: bool) -> io::Result<()> {
1097	self.remote.retire(id, keep)
1098	}
1099	}
1100
1101	#[cfg(test)]
1102	mod tests {
1103	use super::*;
1104	use crate::discover::EntryKind;
1105	const BACKSTOP: Duration = Background::BACKSTOP;
1106	use onestore::{CommitError, Transaction};
1107	use std::{
1108	collections::HashMap,
1109	sync::mpsc::{self, Receiver},
1110	};
1111
1112	fn path(n: usize) -> String {
1113	let folder = if n.is_multiple_of(2) { "" } else { "Group/" };
1114	format!("{folder}Section {n:03}.one")
1115	}
1116
1117	fn listed(n: usize) -> Listed {
1118	Listed {
1119	size: 1000 + n as u64,
1120	modified: 7,
1121	}
1122	}
1123
1124	fn stamp(n: usize) -> Stamp {
1125	Stamp {
1126	header: [n as u8; 1024],
1127	length: n as u64,
1128	}
1129	}
1130
1131	/// `count` sections, each found by discovery as `listed` and `stamp` have it, with `warm`.
1132	fn sections(count: usize, warm: bool) -> Vec<Known> {
1133	(0..count)
1134	.map(|n| Known {
1135	path: path(n),
1136	replica: None,
1137	found: warm.then(|| (listed(n), stamp(n))),
1138	image: None,
1139	})
1140	.collect()
1141	}
1142
1143	#[test]
1144	fn a_notebook_with_every_section_held_still_connects_its_watch() {
1145	let now = Instant::now();
1146	let mut watched = Watched::default();
1147	watched.watch(sections(1, true), now);
1148	let (held, woken) = Signal::new();
1149	watched.sections[0].held = Some(Held {
1150	worker: Arc::downgrade(&held),
1151	replica: Weak::new(),
1152	observed: Weak::new(),
1153	});
1154	assert!(matches!(watched.next(now, false), Next::Connect(0)));
1155	watched.watching(true);
1156	assert!(held.watched.load(Ordering::Acquire));
1157	assert!(matches!(watched.next(now, true), Next::Wait(_)));
1158	assert!(woken.recv_timeout(Duration::from_millis(50)).is_ok());
1159	}
1160
1161	/// The notebook's folders as a listing shows its first `count` sections.
1162	fn listing(count: usize) -> BTreeMap<String, Vec<Entry>> {
1163	let mut folders: BTreeMap<String, Vec<Entry>> = BTreeMap::new();
1164	for n in 0..count {
1165	let path = path(n);
1166	let (folder, name) = split(&path);
1167	folders.entry(folder.to_owned()).or_default().push(Entry {
1168	name: name.to_owned(),
1169	kind: EntryKind::File,
1170	listed: listed(n),
1171	});
1172	}
1173	folders
1174	}
1175
1176	/// Connects at `now`, a watch reporting, and the folders listing as `listings` has them;
1177	/// `meanwhile` runs while they are listed.
1178	fn connect(
1179	watched: &mut Watched,
1180	listings: &BTreeMap<String, Vec<Entry>>,
1181	now: Instant,
1182	meanwhile: impl FnOnce(&mut Watched),
1183	) {
1184	watched.watching(true);
1185	let folders = watched.rescanning();
1186	meanwhile(watched);
1187	let listings = folders
1188	.into_iter()
1189	.map(|folder| (folder.clone(), listings[&folder].clone()))
1190	.collect();
1191	watched.listed(&listings, Some(false), now);
1192	}
1193
1194	/// Runs the schedule as the thread would from `now` to `end`, every check succeeding and
1195	/// every folder listing as `listings` has it, answering each check: when, which section,
1196	/// and whether it read the file's stamp.
1197	fn run(
1198	watched: &mut Watched,
1199	mut now: Instant,
1200	end: Instant,
1201	listings: &BTreeMap<String, Vec<Entry>>,
1202	) -> Vec<(Instant, String, bool)> {
1203	let mut checks = Vec::new();
1204	loop {
1205	match watched.next(now, true) {
1206	Next::Wait(Some(wait)) if now + wait <= end => now += wait,
1207	Next::Wait(_) => return checks,
1208	Next::Connect(_) => unreachable!("connected"),
1209	Next::List(folders) => {
1210	let listings = folders
1211	.into_iter()
1212	.map(|folder| (folder.clone(), listings[&folder].clone()))
1213	.collect();
1214	watched.listed(&listings, None, now);
1215	}
1216	Next::Check { path, current, .. } => {
1217	let interval = watched.interval();
1218	let watch = watched
1219	.sections
1220	.iter_mut()
1221	.find(|watch| watch.path == path)
1222	.unwrap();
1223	watch.current = false;
1224	watch.due = watch.due.min(now + interval);
1225	checks.push((now, path, !current));
1226	}
1227	}
1228	}
1229	}
1230
1231	#[test]
1232	fn a_new_notebook_is_read_once_staggered_then_left_alone() {
1233	let start = Instant::now();
1234	let mut watched = Watched::default();
1235	watched.watch(sections(200, false), start);
1236	connect(&mut watched, &listing(200), start, |_| {});
1237	let checks = run(
1238	&mut watched,
1239	start,
1240	start + BACKSTOP - SETTLE,
1241	&listing(200),
1242	);
1243	assert_eq!(checks.len(), 200, "one first check each, then none");
1244	assert!(checks.iter().all(|(.., read)| *read));
1245	for pair in checks.windows(2) {
1246	assert!(pair[1].0 - pair[0].0 >= STAGGER, "never a burst");
1247	}
1248	// Then each section costs one backstop check an interval: over a day, 24 stamp reads
1249	// where a 15-second poll made 5,760.
1250	let day = run(
1251	&mut watched,
1252	start + BACKSTOP - SETTLE,
1253	start + Duration::from_secs(86_400) - SETTLE,
1254	&listing(200),
1255	);
1256	assert_eq!(day.len(), 200 * (86_400 / BACKSTOP.as_secs() as usize - 1));
1257	}
1258
1259	#[test]
1260	fn reopening_reads_only_the_files_listed_otherwise() {
1261	let start = Instant::now();
1262	let mut watched = Watched::default();
1263	watched.watch(sections(200, true), start);
1264	let mut listings = listing(200);
1265	// Another client wrote one section while the notebook was closed.
1266	listings.get_mut("Group").unwrap()[3].listed.modified += 1;
1267	connect(&mut watched, &listings, start, |_| {});
1268	let checks = run(&mut watched, start, start + SETTLE, &listings);
1269	assert_eq!(checks.len(), 200);
1270	let read: Vec<_> = checks.iter().filter(|(.., read)| *read).collect();
1271	assert_eq!(read.len(), 1, "{read:?}");
1272	assert_eq!(read[0].1, "Group/Section 007.one");
1273	assert!(
1274	checks.iter().all(|(at, ..)| *at == start),
1275	"nothing to read, nothing to stagger"
1276	);
1277	}
1278
1279	#[test]
1280	fn a_change_reported_while_listing_is_read_though_the_listing_missed_it() {
1281	let start = Instant::now();
1282	let mut watched = Watched::default();
1283	watched.watch(sections(4, true), start);
1284	connect(&mut watched, &listing(4), start, |watched| {
1285	watched.touched(&[path(2)], start);
1286	});
1287	let checks = run(&mut watched, start, start + SETTLE * 2, &listing(4));
1288	let read: Vec<_> = checks
1289	.into_iter()
1290	.filter(|(.., read)| *read)
1291	.map(|(_, path, _)| path)
1292	.collect();
1293	assert_eq!(read, vec![path(2)]);
1294	}
1295
1296	#[test]
1297	fn a_reported_change_checks_exactly_the_sections_it_names() {
1298	let start = Instant::now();
1299	let mut watched = Watched::default();
1300	watched.watch(sections(200, true), start);
1301	let mut listings = listing(200);
1302	connect(&mut watched, &listings, start, |_| {});
1303	let now = start + Duration::from_secs(60);
1304	run(&mut watched, start, now, &listings);
1305	// A commit's several writes are reported apiece, and settle into one check.
1306	for _ in 0..3 {
1307	watched.touched(&["section 004.ONE".into()], now);
1308	}
1309	watched.touched(
1310	&["Section 004.one".into()],
1311	now + Duration::from_millis(300),
1312	);
1313	let checks = run(&mut watched, now, now + Duration::from_secs(60), &listings);
1314	assert_eq!(
1315	checks,
1316	vec![(now + SETTLE, "Section 004.one".to_owned(), true)]
1317	);
1318	// A folder, as a watch names it when it cannot name the file, is listed, and only the
1319	// section listed otherwise is checked.
1320	let now = now + Duration::from_secs(60);
1321	listings.get_mut("Group").unwrap()[2].listed.size += 1;
1322	watched.touched(&["Group".into()], now);
1323	let group = run(&mut watched, now, now + SETTLE * 2, &listings);
1324	assert_eq!(
1325	group,
1326	vec![(now + SETTLE, "Group/Section 005.one".to_owned(), true)]
1327	);
1328	// `""` lists every folder; nothing listed otherwise, nothing is checked.
1329	watched.touched(&["Grou".into(), String::new()], now + SETTLE * 2);
1330	assert!(run(&mut watched, now + SETTLE * 2, now + SETTLE * 4, &listings).is_empty());
1331	}
1332
1333	/// Each file's image and how many times it was written, by path.
1334	type Images = HashMap<String, (Vec<u8>, u64)>;
1335
1336	/// Section files in memory, each stamp read reported as it happens.
1337	#[derive(Clone)]
1338	struct Files {
1339	images: Arc<Mutex<Images>>,
1340	stamps: mpsc::Sender<String>,
1341	}
1342
1343	impl Files {
1344	fn write(&self, path: &str, image: Vec<u8>) {
1345	let mut images = self.images.lock().unwrap();
1346	let version = images.get(path).map_or(0, |(_, version)| version + 1);
1347	images.insert(path.to_owned(), (image, version));
1348	}
1349
1350	fn list(&self, folder: &str) -> Vec<Entry> {
1351	let images = self.images.lock().unwrap();
1352	images
1353	.iter()
1354	.filter(|(path, _)| split(path).0 == folder)
1355	.map(|(path, (image, version))| Entry {
1356	name: split(path).1.to_owned(),
1357	kind: EntryKind::File,
1358	listed: Listed {
1359	size: image.len() as u64,
1360	modified: *version,
1361	},
1362	})
1363	.collect()
1364	}
1365
1366	fn known(&self, path: &str) -> Known {
1367	let (image, version) = self.images.lock().unwrap()[path].clone();
1368	Known {
1369	path: path.to_owned(),
1370	replica: None,
1371	found: Some((
1372	Listed {
1373	size: image.len() as u64,
1374	modified: version,
1375	},
1376	Stamp::of(&image).unwrap(),
1377	)),
1378	image: None,
1379	}
1380	}
1381	}
1382
1383	struct File(Files, String);
1384
1385	impl Remote for File {
1386	fn read(&mut self) -> io::Result<Vec<u8>> {
1387	let images = self.0.images.lock().unwrap();
1388	Ok(images[&self.1].0.clone())
1389	}
1390
1391	fn stamp(&mut self) -> io::Result<Stamp> {
1392	let _ = self.0.stamps.send(self.1.clone());
1393	Stamp::of(&self.read()?).map_err(io::Error::other)
1394	}
1395
1396	fn publish(&mut self, _: &Transaction) -> std::result::Result<(), CommitError> {
1397	unreachable!("nothing is queued")
1398	}
1399
1400	fn confirm(&mut self, _: &Stamp) -> std::result::Result<(), CommitError> {
1401	unreachable!("nothing is queued")
1402	}
1403	}
1404
1405	fn quiet(stamps: &Receiver<String>, wait: Duration) -> Vec<String> {
1406	let mut read = Vec::new();
1407	while let Ok(path) = stamps.recv_timeout(wait) {
1408	read.push(path);
1409	}
1410	read
1411	}
1412
1413	#[test]
1414	fn a_watch_wakes_the_section_it_reports_and_a_lost_one_lists_again() {
1415	let (sender, stamps) = mpsc::channel();
1416	let files = Files {
1417	images: Arc::default(),
1418	stamps: sender,
1419	};
1420	let mut paths: Vec<String> = (0..6).map(path).collect();
1421	paths.sort();
1422	for path in &paths {
1423	files.write(
1424	path,
1425	onestore::create_section("s.one", path, "Author").unwrap(),
1426	);
1427	}
1428	let (armed, watches) = mpsc::channel();
1429	let remote = files.clone();
1430	let background = Background::start(
1431	false,
1432	move |reports| {
1433	armed.send(reports).unwrap();
1434	let (files, listed) = (remote.clone(), remote.clone());
1435	let bind = move |path: &str| File(files.clone(), path.to_owned());
1436	let list = move |folder: String| std::future::ready(Ok(listed.list(&folder)));
1437	Ok(((bind, list), true))
1438	},
1439	|| {},
1440	)
1441	.unwrap();
1442	// A section the catalog read unchanged is not read again; a new one is.
1443	let mut known: Vec<Known> = paths[1..].iter().map(|path| files.known(path)).collect();
1444	known.push(Known {
1445	path: paths[0].clone(),
1446	replica: None,
1447	found: None,
1448	image: None,
1449	});
1450	background.watch(known);
1451	assert_eq!(quiet(&stamps, STAGGER * 4), vec![paths[0].clone()]);
1452	let watch = watches.recv().unwrap();
1453	assert!(quiet(&stamps, SETTLE).is_empty(), "idle reads nothing");
1454
1455	// A folder reported without its file is listed; only the file listed otherwise is read.
1456	let other = &paths[4];
1457	files.write(
1458	other,
1459	onestore::create_section("s.one", "Other", "Author").unwrap(),
1460	);
1461	watch.touched(&[String::new()]);
1462	assert_eq!(quiet(&stamps, SETTLE * 2), vec![other.clone()]);
1463	assert_eq!(background.changed(), vec![other.clone()]);
1464
1465	let changed = &paths[3];
1466	files.write(
1467	changed,
1468	onestore::create_section("s.one", "Changed", "Author").unwrap(),
1469	);
1470	watch.touched(std::slice::from_ref(changed));
1471	assert_eq!(quiet(&stamps, SETTLE * 2), vec![changed.clone()]);
1472	assert_eq!(background.changed(), vec![changed.clone()]);
1473
1474	// A watch that ended leaves changes unreported, so every folder is listed again, and
1475	// the files listed otherwise since the last listing are read.
1476	let third = &paths[5];
1477	files.write(
1478	third,
1479	onestore::create_section("s.one", "Third", "Author").unwrap(),
1480	);
1481	watch.lost();
1482	let rewatch = watches.recv_timeout(SETTLE).unwrap();
1483	let mut read = quiet(&stamps, STAGGER * 4);
1484	read.sort();
1485	assert_eq!(read, vec![changed.clone(), third.clone()]);
1486	// The ended watch's connection is gone; only the current one's loss counts.
1487	watch.lost();
1488	assert!(quiet(&stamps, SETTLE).is_empty());
1489	drop(rewatch);
1490	}
1491
1492	#[test]
1493	fn a_section_released_during_its_check_is_checked_again_at_once() {
1494	let (sender, stamps) = mpsc::channel();
1495	let files = Files {
1496	images: Arc::default(),
1497	stamps: sender,
1498	};
1499	let section = path(0);
1500	files.write(
1501	&section,
1502	onestore::create_section("s.one", &section, "Author").unwrap(),
1503	);
1504	let (armed, watches) = mpsc::channel();
1505	let remote = files.clone();
1506	let background = Background::start(
1507	false,
1508	move |reports| {
1509	armed.send(reports).unwrap();
1510	let (files, listed) = (remote.clone(), remote.clone());
1511	let bind = move |path: &str| File(files.clone(), path.to_owned());
1512	let list = move |folder: String| std::future::ready(Ok(listed.list(&folder)));
1513	Ok(((bind, list), true))
1514	},
1515	|| {},
1516	)
1517	.unwrap();
1518	background.watch(vec![files.known(&section)]);
1519	let watch = watches.recv().unwrap();
1520	assert!(quiet(&stamps, SETTLE).is_empty());
1521	// Holding the files parks the check reported next inside its stamp read.
1522	let parked = files.images.lock().unwrap();
1523	watch.touched(std::slice::from_ref(&section));
1524	assert_eq!(stamps.recv_timeout(SETTLE * 2).unwrap(), section);
1525	background.0.released(&section);
1526	drop(parked);
1527	assert_eq!(stamps.recv_timeout(SETTLE).unwrap(), section);
1528	drop(background);
1529	}
1530
1531	#[test]
1532	fn held_status_tracks_its_worker_without_retaining_the_replica() {
1533	use crate::session::SyncState;
1534	use onestore::{
1535	PageCreation,
1536	op::{Edit, Op, SectionOp},
1537	};
1538
1539	let directory = tempfile::tempdir().unwrap();
1540	let file = directory.path().join("Held.one");
1541	let source = onestore::create_section("Held.one", "", "Author").unwrap();
1542	fs::write(&file, &source).unwrap();
1543	let (changed, changes) = mpsc::channel();
1544	let section = Section::open(&file, directory.path().join("cache"), false, move || {
1545	let _ = changed.send(());
1546	})
1547	.unwrap();
1548	changes.recv_timeout(Duration::from_secs(3)).unwrap();
1549	section.set_offline(true);
1550
1551	let (stamps, _) = mpsc::channel();
1552	let files = Files {
1553	images: Arc::default(),
1554	stamps,
1555	};
1556	files.write("Held.one", source);
1557	let remote = files.clone();
1558	let background = Background::start(
1559	false,
1560	move |_| {
1561	let (files, listed) = (remote.clone(), remote.clone());
1562	let bind = move |path: &str| File(files.clone(), path.to_owned());
1563	let list = move |folder: String| std::future::ready(Ok(listed.list(&folder)));
1564	Ok(((bind, list), true))
1565	},
1566	|| {},
1567	)
1568	.unwrap();
1569	background.set_offline(true);
1570	background.watch(vec![files.known("Held.one")]);
1571	background.hold("Held.one", &section);
1572	section
1573	.apply(
1574	"Author",
1575	Edit {
1576	at: crate::now(),
1577	ops: vec![Op::Section(SectionOp::Create(
1578	PageCreation::new(None, Some("Queued"), "Author").unwrap(),
1579	))],
1580	},
1581	)
1582	.unwrap();
1583	section.written().unwrap();
1584	let pending = section.pending().unwrap();
1585	assert_eq!(pending.len(), 1);
1586	let status = background.status().remove(0).1;
1587	assert_eq!(status.queued, 1);
1588	assert!(status.synced.is_some());
1589	assert_eq!(status.state(), SyncState::Syncing);
1590	{
1591	let watched = background.0.watched.lock().unwrap();
1592	assert!(watched.sections[0].status.synced.is_none());
1593	assert_eq!(watched.sections[0].status.queued, 0);
1594	}
1595
1596	let wait = |expected| {
1597	let deadline = Instant::now() + Duration::from_secs(3);
1598	loop {
1599	let status = background.status().remove(0).1;
1600	if status.state() == expected {
1601	return status;
1602	}
1603	changes
1604	.recv_timeout(deadline.saturating_duration_since(Instant::now()))
1605	.unwrap();
1606	}
1607	};
1608	section.set_offline(false);
1609	assert_eq!(wait(SyncState::UpToDate).queued, 0);
1610	assert!(matches!(
1611	section.status(pending[0].id).unwrap(),
1612	Some(EditStatus::Published { .. })
1613	));
1614
1615	let published = fs::read_file(&file).unwrap();
1616	fs::remove_file(&file).unwrap();
1617	section.wake();
1618	assert_eq!(
1619	wait(SyncState::NotConnected).error.unwrap().kind(),
1620	io::ErrorKind::NotFound
1621	);
1622	fs::write(&file, published).unwrap();
1623	section.wake();
1624	assert!(wait(SyncState::UpToDate).error.is_none());
1625	assert!(
1626	background.0.watched.lock().unwrap().sections[0]
1627	.status
1628	.synced
1629	.is_none()
1630	);
1631
1632	let replica = Arc::downgrade(section.replica());
1633	section.close().unwrap();
1634	assert!(replica.upgrade().is_none());
1635	assert!(
1636	background.0.watched.lock().unwrap().sections[0]
1637	.held
1638	.is_none()
1639	);
1640	}
1641
1642	#[test]
1643	fn a_held_section_is_left_to_its_worker_which_the_watch_wakes() {
1644	let (sender, stamps) = mpsc::channel();
1645	let files = Files {
1646	images: Arc::default(),
1647	stamps: sender,
1648	};
1649	let held = path(0);
1650	files.write(
1651	&held,
1652	onestore::create_section("s.one", &held, "Author").unwrap(),
1653	);
1654	let (armed, watches) = mpsc::channel();
1655	let remote = files.clone();
1656	let background = Background::start(
1657	false,
1658	move |reports| {
1659	armed.send(reports).unwrap();
1660	let (files, listed) = (remote.clone(), remote.clone());
1661	let bind = move |path: &str| File(files.clone(), path.to_owned());
1662	let list = move |folder: String| std::future::ready(Ok(listed.list(&folder)));
1663	Ok(((bind, list), true))
1664	},
1665	|| {},
1666	)
1667	.unwrap();
1668	background.watch(vec![files.known(&held)]);
1669	let watch = watches.recv().unwrap();
1670	assert!(quiet(&stamps, SETTLE).is_empty());
1671	// As `hold` leaves it to a session's worker.
1672	let (signal, woken) = Signal::new();
1673	if let Ok(mut watched) = background.0.watched.lock() {
1674	watched.sections[0].held = Some(Held {
1675	worker: Arc::downgrade(&signal),
1676	replica: Weak::new(),
1677	observed: Weak::new(),
1678	});
1679	signal.watched.store(watched.reported, Ordering::Release);
1680	}
1681	assert!(
1682	signal.watched.load(Ordering::Acquire),
1683	"the watch reports, so the worker need not poll"
1684	);
1685	watch.touched(std::slice::from_ref(&held));
1686	assert!(woken.recv_timeout(SETTLE * 2).is_ok(), "the watch wakes it");
1687	assert!(
1688	quiet(&stamps, SETTLE).is_empty(),
1689	"the worker reads, not this"
1690	);
1691	watch.lost();
1692	assert!(woken.recv_timeout(SETTLE).is_ok());
1693	let rearmed = watches.recv_timeout(SETTLE).unwrap();
1694	while !signal.watched.load(Ordering::Acquire) {
1695	woken.recv_timeout(SETTLE * 2).unwrap();
1696	}
1697	assert!(signal.watched.load(Ordering::Acquire));
1698	rearmed.touched(std::slice::from_ref(&held));
1699	assert!(woken.recv_timeout(SETTLE * 2).is_ok());
1700	assert!(quiet(&stamps, SETTLE).is_empty());
1701	drop(signal);
1702	drop(background);
1703	}
1704	}