Commit c345a48e944a431673a0e8b9d4a59e93dc507a58 | snowbound

author	clover caruso <git@paperclover.net> 2026-10-15 13:00:06-07:00
committer	clover caruso <git@paperclover.net> 2026-10-15 15:21:41-07:00
log	c345a48e944a431673a0e8b9d4a59e93dc507a58
tree	50a0eb26f02cec8f68a9cba6c15d643ee3277e09
parent	c780bc064c84cc299b01189b7b225c335dd00da7
signature	Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU (~clover)

feat(notebook): merge publications a provider lost

A document provider has no lock or compare-and-swap, so another device's upload can replace this one's and keep whichever lands last. Each section now keeps the images it publishes in the cache until the document is seen holding them, and a publication that was lost merges back into the document as an iCloud conflict version does, with conflict pages where edits clash, named for this device. A document still standing as it was before an upload gets a minute for reads to catch up before the upload counts as lost, and an upload a provider writing in place left torn is put back from the image this device kept.

With lost publications protected, sections a provider only streams are no longer read-only: their edits publish by writing the whole section again. On Android the notebook passes the device's name along.

Refs #24

Assisted-by: claude-opus-5.5

11 files changed, 816 insertions(+), 164 deletions(-)

apps
- android
  - jni
    snowbound.c
  - src
    net
    paperclover
    snowbound
    Editor.java
crates
- mobile
  - include
    snowbound.h
  - src
    library.rs
    provider.rs
    tests.rs
- notebook
  - README.md
  - src
    provider.rs
    provider
    tests.rs
    session.rs
    sync.rs

apps/android/jni/snowbound.c+5-4

...	...	@@ -247,12 +247,13 @@ JNIEXPORT void JNICALL PROVIDER(renamed)(JNIEnv *env, jclass cls, jlong renamed,
247	247	}
248	248
249	249	JNIEXPORT jlong JNICALL JNI(provided)(JNIEnv *env, jclass cls, jstring root, jboolean folder, jboolean local,
250		jstring cache) {
	250	jstring device, jstring cache) {
251	251	(void)cls;
252		char r = utf8(env, root), c = r ? utf8(env, cache) : NULL, *error = NULL;
253		const Library *library = c ? sb_library_provider(r, folder, local, c, &error) : NULL;
	252	char r = utf8(env, root), d = r ? utf8(env, device) : NULL, *c = d ? utf8(env, cache) : NULL;
	253	char *error = NULL;
	254	const Library *library = c ? sb_library_provider(r, folder, local, d, c, &error) : NULL;
254	255	if (c && !library) fail(env, error, "The notebook can't be opened");
255		free(r), free(c);
	256	free(r), free(d), free(c);
256	257	return (jlong)(intptr_t)library;
257	258	}
258	259

apps/android/src/net/paperclover/snowbound/Editor.java+11-2

...	...	@@ -1,9 +1,11 @@
1	1	package net.paperclover.snowbound;
2	2
3	3	import android.content.Context;
	4	import android.os.Build;
4	5	import android.os.Handler;
5	6	import android.os.Looper;
6	7	import android.util.Log;
	8	import android.provider.Settings;
7	9	import android.view.Surface;
8	10	import java.io.File;
9	11	import java.text.DateFormat;
...	...	@@ -62,7 +64,8 @@ final class Editor {
62	64	if (picked != null && granted) {
63	65	Provider.writable = picked.writable();
64	66	try {
65		opened = provided(picked.root(), picked.folder, picked.local(), cache.getPath());
	67	opened = provided(picked.root(), picked.folder, picked.local(), device(context),
	68	cache.getPath());
66	69	path = firstSection(opened);
67	70	first = section(opened, path);
68	71	} catch (IllegalStateException \| JSONException error) {
...	...	@@ -139,6 +142,12 @@ final class Editor {
139	142	return false;
140	143	}
141	144
	145	/** The name the user gave this device, as conflict pages name who made them. */
	146	private static String device(Context context) {
	147	String name = Settings.Global.getString(context.getContentResolver(), Settings.Global.DEVICE_NAME);
	148	return name != null ? name : Build.MODEL;
	149	}
	150
142	151	/** Called on the section's thread when it has events. */
143	152	static void woken() {
144	153	MAIN.post(Editor::poll);
...	...	@@ -232,6 +241,6 @@ final class Editor {
232	241	static native void reload(long view, boolean discard);
233	242	static native void freeView(long view);
234	243	static native boolean close(long library, long section, double seconds);
235		static native long provided(String root, boolean folder, boolean local, String cache);
	244	static native long provided(String root, boolean folder, boolean local, String device, String cache);
236	245	static native String syncStatus(long library);
237	246	}

crates/mobile/include/snowbound.h+3-2

...	...	@@ -39,7 +39,7 @@ typedef void (sb_document_found)(void context, const char id, const char nam
39	39	typedef struct sb_provider {
40	40	int32_t (document)(const char id, sb_document_found found, void *context);
41	41	int32_t (children)(const char id, sb_document_found found, void *context);
42		// mode is "r" or "rw"; fd is a duplicate Snowbound owns and closes, handle the descriptor
	42	// mode is "r", "rw", "wt" or "w"; fd is a duplicate Snowbound owns and closes, handle the descriptor
43	43	// the host keeps open: detachFd would end the channel checkError reads. A nonzero status
44	44	// sets neither; a negative fd with status 0 has its handle closed.
45	45	int32_t (open)(const char id, const char mode, int32_t fd, uint64_t *handle);
...	...	@@ -50,7 +50,8 @@ typedef struct sb_provider {
50	50	void sb_set_provider(const sb_provider *provider);
51	51
52	52	const Library sb_library_open(const char path, const char cache, bool local, char *error);
53		const Library sb_library_provider(const char root, bool folder, bool local, const char cache, char *error);
	53	const Library sb_library_provider(const char root, bool folder, bool local, const char device, const char cache,
	54	char **error);
54	55	const Library sb_library_server(const char address, const char share, const char user, const char *password,
55	56	const char domain, const char root, const char cache, char *error);
56	57	// Callback/token live until the blocking join returns; false cancels, approving reports host approval.

crates/mobile/src/library.rs+16-7

...	...	@@ -512,7 +512,7 @@ impl Library {
512	512	/// registered provider keeps (`sb_set_provider`). A provider `local` to this device
513	513	/// publishes each edit at once; any other waits for a pause in typing. While the provider
514	514	/// can't be reached, the folder opens as last listed and sections from their offline copies.
515		fn provider(root: &str, folder: bool, local: bool, cache: &Path) -> Result<Self> {
	515	fn provider(root: &str, folder: bool, local: bool, device: &str, cache: &Path) -> Result<Self> {
516	516	let host = crate::provider::Host::default();
517	517	let name = (!folder)
518	518	.then(\|\| {
...	...	@@ -523,9 +523,9 @@ impl Library {
523	523	)
524	524	})
525	525	.flatten();
526		let documents = Arc::new(Documents::new(host, root));
	526	let documents = Arc::new(Documents::new(host, root, cache, device));
527	527	let (notebook, place, background) = if folder {
528		let mut notebook = Notebook::open_provider(Arc::clone(&documents), cache)?;
	528	let mut notebook = Notebook::open_provider(Arc::clone(&documents))?;
529	529	let background = Background::provider(Arc::clone(&documents), sync_woken)?;
530	530	background.watch(notebook.replicas());
531	531	(Some(notebook), Place::Provider(documents), Some(background))
...	...	@@ -824,7 +824,6 @@ impl Library {
824	824	)?),
825	825	Place::Document(documents, _) => Ok(session::Section::open_provider(
826	826	Arc::clone(documents),
827		&self.cache,
828	827	offline,
829	828	notify,
830	829	)?),
...	...	@@ -2224,22 +2223,32 @@ pub unsafe extern "C" fn sb_library_flush(
2224	2223
2225	2224	/// The notebook folder whose document is `root`, or with `folder` false the lone section
2226	2225	/// `root` names, as the provider `sb_set_provider` registered keeps it; `local` as
2227		/// `sb_library_open` takes it. While the provider can't be reached, a folder listed before
	2226	/// `sb_library_open` takes it, and `device` naming this device on the conflict pages its own
	2227	/// edits make when another device's upload replaced them. While the provider can't be reached, a folder listed before
2228	2228	/// opens as last listed, and sections opened before open from their copies; a revoked grant
2229	2229	/// refuses to open and keeps them.
2230	2230	///
2231	2231	/// # Safety
2232		/// `root` and `cache` are NUL-terminated UTF-8; `error` is null or a place for a string.
	2232	/// `root`, `device` and `cache` are NUL-terminated UTF-8; `error` is null or a place for a
	2233	/// string.
2233	2234	#[unsafe(no_mangle)]
2234	2235	pub unsafe extern "C" fn sb_library_provider(
2235	2236	root: *const c_char,
2236	2237	folder: bool,
2237	2238	local: bool,
	2239	device: *const c_char,
2238	2240	cache: *const c_char,
2239	2241	error: mut mut c_char,
2240	2242	) -> *const Library {
	2243	let cache = string(cache);
2241	2244	shared(
2242		Library::provider(&string(root), folder, local, Path::new(&string(cache))),
	2245	Library::provider(
	2246	&string(root),
	2247	folder,
	2248	local,
	2249	&string(device),
	2250	Path::new(&cache),
	2251	),
2243	2252	error,
2244	2253	)
2245	2254	}

crates/mobile/src/provider.rs+1-1

...	...	@@ -35,7 +35,7 @@ pub struct Table {
35	35	pub document: extern "C" fn(id: const c_char, found: Found, context: mut c_void) -> i32,
36	36	/// Calls `found` for each document in the folder `id`.
37	37	pub children: extern "C" fn(id: const c_char, found: Found, context: mut c_void) -> i32,
38		/// Opens `id` in `ContentResolver` mode `r` or `rw`, setting `fd` to a descriptor the
	38	/// Opens `id` in a `ContentResolver` mode, setting `fd` to a descriptor the
39	39	/// caller owns and closes, and `handle` to what `close` finishes.
40	40	pub open: extern "C" fn(
41	41	id: *const c_char,

crates/mobile/src/tests.rs+9-2

...	...	@@ -3752,9 +3752,16 @@ fn provided(root: &Path, folder: bool, cache: &Path) -> std::result::Result<Arc<
3752	3752	CString::new(cache.to_str().unwrap()).unwrap(),
3753	3753	);
3754	3754	let mut error = std::ptr::null_mut();
3755		// SAFETY: both strings are NUL-terminated and `error` a place for one.
	3755	// SAFETY: every string is NUL-terminated and `error` a place for one.
3756	3756	let library = unsafe {
3757		library::sb_library_provider(root.as_ptr(), folder, true, cache.as_ptr(), &mut error)
	3757	library::sb_library_provider(
	3758	root.as_ptr(),
	3759	folder,
	3760	true,
	3761	c"Test".as_ptr(),
	3762	cache.as_ptr(),
	3763	&mut error,
	3764	)
3758	3765	};
3759	3766	if library.is_null() {
3760	3767	let message = string(error);

crates/notebook/README.md+20-7

...	...	@@ -551,11 +551,11 @@ physical power-loss durability.
551	551
552	552	`provider` opens notebooks a document provider keeps, as Android's Storage Access Framework
553	553	reaches them. The host implements `provider::Provider` (a document's description, a folder's
554		children, `open` in `ContentResolver`'s `r` or `rw` as an owned `File`, `close` with the provider's
	554	children, `open` in a `ContentResolver` mode as an owned `File`, `close` with the provider's
555	555	verdict, `rename` returning the new id) and reports an unreachable provider as `NotConnected`,
556	556	a revoked grant as `PermissionDenied` and a gone document as `NotFound`.
557		`provider::Documents::new(provider, root)` holds one root: a picked folder opens
558		with `Notebook::open_provider(documents, cache)`, its sections with `Section::resume_provider`
	557	`provider::Documents::new(provider, root, cache, device)` holds one root, the app's
	558	cache and the device's name: a picked folder opens with `Notebook::open_provider(documents)`, its sections with `Section::resume_provider`
559	559	and the rest with `Background::provider`, as a Live Share guest's do; a picked document opens
560	560	as a lone section with `Section::open_provider`. Catalog paths resolve to ids by listing their
561	561	folders, again once when an id has gone, as a rename leaves it; of documents sharing a name the
...	...	@@ -564,10 +564,23 @@ first by id lists and the rest are `unavailable`, named with their ids.
564	564	A publication resolves the document's id once, from the last listing, and writes nothing to
565	565	a document that took its name since. Without a write grant a document is only read, and
566	566	publication is refused as `ReadOnlyFilesystem`; where its `r` descriptor seeks, `rw` opens it
567		and a commit checks the stamp and appends in place exactly as on a disk. A document that only
568		streams is read whole for each stamp and is read-only too, its edits waiting, until lost
569		publications merge back. No provider offers a lock or compare-and-swap: the check sees only
570		this device's view of the document. An image that does not parse is never a base. An open
	567	and a commit checks the stamp and appends in place exactly as on a disk; a document that only
	568	streams is read whole for each stamp, and a commit reads it, checks its stamp, applies the
	569	transaction and writes the whole image with `wt`, or `w` where that is refused, images only
	570	growing. No provider offers a lock or compare-and-swap: the check sees only this device's view
	571	of the document, and another device's upload can replace this one's. So each image a section
	572	publishes is kept in the cache, beside its replica: once written in place, as the image it was
	573	published on (kept the same way) with the transaction applied, else read back; once a whole
	574	upload has opened, as uploaded. Once the document is seen otherwise it is listed as a
	575	`Remote::versions` version, until the document holds the image's revisions: a lost publication
	576	merges back as a conflict version does, conflict pages where it clashes, named for the device
	577	`Documents::new` was given. A document still standing as the image an upload was made on is
	578	given a minute for reads to catch up before the upload counts as lost. Only this device's
	579	publications are kept so, and one that cannot merge is set aside in the cache, as no new
	580	document can be made beside it. A document read torn as this device's own upload cut off on a
	581	provider that writes in place leaves it, the start of an image it kept, is written again from
	582	that image, once a step; any other image that does not parse is left as it is, and is never a
	583	base. An open
571	584	section checks its file every `Background::UNWATCHED`, the host's change reports waking it
572	585	between. While the provider can't be reached the notebook opens as its folders were last
573	586	listed and sections from their replicas; a revoked grant keeps them, and the notebook does not

crates/notebook/src/provider.rs+342-50

...	...	@@ -12,6 +12,7 @@ use std::{
12	12	io::{self, Read, Seek, SeekFrom, Write},
13	13	path::PathBuf,
14	14	sync::{Arc, Mutex, MutexGuard, PoisonError},
	15	time::Duration,
15	16	};
16	17
17	18	/// A document as its provider describes it.
...	...	@@ -33,7 +34,7 @@ pub struct Document {
33	34	pub trait Provider: Send + Sync {
34	35	fn document(&self, id: &str) -> io::Result<Document>;
35	36	fn children(&self, id: &str) -> io::Result<Vec<Document>>;
36		/// Opens a document in a `ContentResolver` mode, `r` or `rw`.
	37	/// Opens a document in a `ContentResolver` mode: `r`, `rw`, `wt` or `w`.
37	38	fn open(&self, id: &str, mode: &str) -> io::Result<File>;
38	39	/// Closes a file `open` gave, with the provider's verdict on what was written to it.
39	40	fn close(&self, file: File) -> io::Result<()>;
...	...	@@ -46,15 +47,26 @@ pub trait Provider: Send + Sync {
46	47	pub struct Documents {
47	48	provider: Box<dyn Provider>,
48	49	pub(crate) root: String,
	50	/// The app's cache, holding the replicas and the notebook's listing.
	51	pub(crate) cache: PathBuf,
	52	/// This device as the conflict pages its merged-back publications make name it.
	53	device: String,
49	54	/// Each path's document as last listed; a stale one is listed again.
50	55	ids: Mutex<BTreeMap<String, String>>,
51	56	}
52	57
53	58	impl Documents {
54		pub fn new(provider: impl Provider + 'static, root: impl Into<String>) -> Self {
	59	pub fn new(
	60	provider: impl Provider + 'static,
	61	root: impl Into<String>,
	62	cache: impl Into<PathBuf>,
	63	device: impl Into<String>,
	64	) -> Self {
55	65	Self {
56	66	provider: Box::new(provider),
57	67	root: root.into(),
	68	cache: cache.into(),
	69	device: device.into(),
58	70	ids: Mutex::default(),
59	71	}
60	72	}
...	...	@@ -63,6 +75,12 @@ impl Documents {
63	75	format!("provider:{}", self.root)
64	76	}
65	77
	78	/// Where this device keeps the images it published until the documents are seen holding
	79	/// them: a provider without compare-and-swap may let another device's upload replace one.
	80	fn published(&self) -> PathBuf {
	81	crate::location::folder(&self.cache, &self.location()).join("published")
	82	}
	83
66	84	fn id(&self, path: &str) -> io::Result<String> {
67	85	if path.is_empty() {
68	86	return Ok(self.root.clone());
...	...	@@ -150,9 +168,12 @@ impl Documents {
150	168	self.opened(id, "r", \|file\| {
151	169	if file.stream_position().is_err() {
152	170	let image = whole(file, crate::MAX_FILE_BYTES)?;
153		let stamp = Stamp::of(&image)
154		.map_err(\|error\| io::Error::new(io::ErrorKind::InvalidData, error))?;
155		return Ok((stamp, Some(image)));
	171	// A torn image shorter than a header still stamps, as it stands.
	172	let mut header = [0; 1024];
	173	let start = &image[..image.len().min(1024)];
	174	header[..start.len()].copy_from_slice(start);
	175	let length = image.len() as u64;
	176	return Ok((Stamp { header, length }, Some(image)));
156	177	}
157	178	let mut header = [0; 1024];
158	179	file.read_exact(&mut header)?;
...	...	@@ -162,42 +183,80 @@ impl Documents {
162	183	})
163	184	}
164	185
165		/// Writes the document at `path` in place through `operation`, provided its grant allows
166		/// writing and it opens seekable. The id is the one the last listing gave, never listed
167		/// again, so that a document taking the name meanwhile is not the one written.
	186	/// Writes the document at `path` through `operation`, provided its grant allows writing:
	187	/// in place where its `r` descriptor seeks, else on the image read whole, which then
	188	/// replaces it. `keep` takes what was written: the descriptor, once the commit landed, or
	189	/// the image, once a write to the document opened. The id is the one the last listing gave,
	190	/// never listed again, so that a document taking the name meanwhile is not the one written.
168	191	fn write(
169	192	&self,
170	193	path: &str,
171		operation: impl FnOnce(&mut Seekable) -> std::result::Result<(), CommitError>,
	194	operation: impl FnOnce(&mut Io) -> std::result::Result<(), CommitError>,
	195	keep: impl FnOnce(Written<'_>) -> io::Result<()>,
172	196	) -> std::result::Result<(), CommitError> {
173		let refused = \|error\| CommitError {
174		state: CommitState::NotCommitted,
175		error,
176		};
177		let read_only = \|reason\| refused(io::Error::new(io::ErrorKind::ReadOnlyFilesystem, reason));
178	197	let id = self.id(path).map_err(refused)?;
179		if !self.provider.document(&id).map_err(refused)?.writable {
180		return Err(read_only(
181		"The app keeping this section lets Snowbound only read it",
182		));
183		}
	198	self.writable(&id).map_err(refused)?;
184	199	let seekable = self
185	200	.opened(&id, "r", \|file\| Ok(file.stream_position().is_ok()))
186	201	.map_err(refused)?;
187		// Streamed documents stay read-only until lost publications merge back (plan commit 2).
188		if !seekable {
189		return Err(read_only(
190		"The app keeping this section can't save changes in place yet; they wait on this device",
191		));
192		}
193		let mut io = Seekable(self.provider.open(&id, "rw").map_err(refused)?);
	202	let mut io = if seekable {
	203	Io::File(self.provider.open(&id, "rw").map_err(refused)?)
	204	} else {
	205	let image = self.opened(&id, "r", \|file\| whole(file, crate::MAX_FILE_BYTES));
	206	Io::Image(image.map_err(refused)?)
	207	};
194	208	let result = operation(&mut io);
195		let closed = self.provider.close(io.0);
196		result?;
197		closed.map_err(\|error\| CommitError {
198		state: CommitState::Unknown,
199		error,
200		})
	209	match io {
	210	Io::File(mut file) => {
	211	// A copy it could not keep only leaves the publication unprotected.
	212	if result.is_ok() {
	213	let _ = keep(Written::InPlace(&mut file));
	214	}
	215	let closed = self.provider.close(file);
	216	result?;
	217	closed.map_err(uncertain)
	218	}
	219	Io::Image(image) => {
	220	result?;
	221	self.upload(&id, &image, keep)
	222	}
	223	}
	224	}
	225
	226	/// Fails unless the grant allows writing the document `id`.
	227	fn writable(&self, id: &str) -> io::Result<()> {
	228	if self.provider.document(id)?.writable {
	229	return Ok(());
	230	}
	231	Err(io::Error::new(
	232	io::ErrorKind::ReadOnlyFilesystem,
	233	"The app keeping this section lets Snowbound only read it",
	234	))
	235	}
	236
	237	/// Writes `image` whole as the document `id`, with `wt`, or `w` where that is refused
	238	/// (images only grow), having `keep` take it once a write opened.
	239	fn upload(
	240	&self,
	241	id: &str,
	242	image: &[u8],
	243	keep: impl FnOnce(Written<'_>) -> io::Result<()>,
	244	) -> std::result::Result<(), CommitError> {
	245	// A provider refusing `wt` may say so as any failure but a revoked grant.
	246	let mut file = match self.provider.open(id, "wt") {
	247	Err(error) if error.kind() != io::ErrorKind::PermissionDenied => {
	248	self.provider.open(id, "w")
	249	}
	250	opened => opened,
	251	}
	252	.map_err(refused)?;
	253	if let Err(error) = keep(Written::Image(image)) {
	254	let _ = self.provider.close(file);
	255	return Err(refused(error));
	256	}
	257	let written = file.write_all(image);
	258	let closed = self.provider.close(file);
	259	written.and(closed).map_err(uncertain)
201	260	}
202	261
203	262	/// Renames the document at `path` within its folder.
...	...	@@ -208,6 +267,20 @@ impl Documents {
208	267	}
209	268	}
210	269
	270	fn refused(error: io::Error) -> CommitError {
	271	CommitError {
	272	state: CommitState::NotCommitted,
	273	error,
	274	}
	275	}
	276
	277	fn uncertain(error: io::Error) -> CommitError {
	278	CommitError {
	279	state: CommitState::Unknown,
	280	error,
	281	}
	282	}
	283
211	284	fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
212	285	mutex.lock().unwrap_or_else(PoisonError::into_inner)
213	286	}
...	...	@@ -221,30 +294,116 @@ fn whole(file: &mut File, limit: usize) -> io::Result<Vec<u8>> {
221	294	Ok(bytes)
222	295	}
223	296
224		/// A seekable descriptor as a commit writes it.
225		struct Seekable(File);
	297	/// What a write wrote: the document in place, or its whole image.
	298	enum Written<'a> {
	299	InPlace(&'a mut File),
	300	Image(&'a [u8]),
	301	}
226	302
227		impl CommitIo for Seekable {
	303	/// A document as a commit writes it: its seekable descriptor, or its image to write whole.
	304	enum Io {
	305	File(File),
	306	Image(Vec<u8>),
	307	}
	308
	309	impl CommitIo for Io {
228	310	fn read_at(&mut self, offset: u64, bytes: &mut [u8]) -> io::Result<usize> {
229		self.0.seek(SeekFrom::Start(offset))?;
230		self.0.read(bytes)
	311	match self {
	312	Io::File(file) => {
	313	file.seek(SeekFrom::Start(offset))?;
	314	file.read(bytes)
	315	}
	316	Io::Image(image) => {
	317	let rest = usize::try_from(offset)
	318	.ok()
	319	.and_then(\|offset\| image.get(offset..))
	320	.unwrap_or_default();
	321	let count = rest.len().min(bytes.len());
	322	bytes[..count].copy_from_slice(&rest[..count]);
	323	Ok(count)
	324	}
	325	}
231	326	}
232	327
233	328	fn write_at(&mut self, offset: u64, bytes: &[u8]) -> io::Result<usize> {
234		self.0.seek(SeekFrom::Start(offset))?;
235		self.0.write(bytes)
	329	match self {
	330	Io::File(file) => {
	331	file.seek(SeekFrom::Start(offset))?;
	332	file.write(bytes)
	333	}
	334	Io::Image(image) => {
	335	let offset = usize::try_from(offset).map_err(\|_\| io::ErrorKind::FileTooLarge)?;
	336	let end = offset + bytes.len();
	337	if image.len() < end {
	338	image.resize(end, 0);
	339	}
	340	image[offset..end].copy_from_slice(bytes);
	341	Ok(bytes.len())
	342	}
	343	}
236	344	}
237	345
238	346	fn flush(&mut self) -> io::Result<()> {
239		self.0.sync_all()
	347	match self {
	348	Io::File(file) => file.sync_all(),
	349	Io::Image(_) => Ok(()),
	350	}
240	351	}
241	352	}
242	353
243		/// A section a document provider keeps, as its replica publishes to it.
	354	/// The hex of a stamp's digest, which names an image this device keeps.
	355	fn digest(stamp: &Stamp) -> String {
	356	use sha2::Digest;
	357	let digest = sha2::Sha256::new()
	358	.chain_update(stamp.header)
	359	.chain_update(stamp.length.to_le_bytes())
	360	.finalize();
	361	hex(&digest[..16])
	362	}
	363
	364	fn hex(bytes: &[u8]) -> String {
	365	bytes.iter().map(\|byte\| format!("{byte:02x}")).collect()
	366	}
	367
	368	fn file_of(header: &[u8]) -> Option<[u8; 16]> {
	369	Some(onestore::Header::parse(header).ok()?.file_id)
	370	}
	371
	372	/// An image this device keeps, as its name gives it: `<document>-<image>-<base>.one`, the
	373	/// document's identity, then the digests of the image's stamp and of the stamp of the image
	374	/// it was published on.
	375	struct Kept {
	376	name: String,
	377	document: String,
	378	image: String,
	379	base: String,
	380	}
	381
	382	impl Kept {
	383	fn named(name: String) -> Option<Self> {
	384	let mut parts = name.strip_suffix(".one")?.split('-').map(str::to_owned);
	385	let (document, image, base) = (parts.next()?, parts.next()?, parts.next()?);
	386	Some(Self {
	387	name,
	388	document,
	389	image,
	390	base,
	391	})
	392	}
	393	}
	394
	395	/// How long reads may lag this device's own upload before a document still standing as the
	396	/// image the upload was made on counts as having lost it.
	397	const LAG: Duration = Duration::from_secs(60);
	398
	399	/// A section a document provider keeps, as its replica publishes to it. Each image it
	400	/// publishes is kept, and listed as a version once the document is seen otherwise, until the
	401	/// document holds it: an upload another device's replaced then merges back as a conflict
	402	/// version does (`resolve.rs`). Only this device's publications are kept so.
244	403	pub(crate) struct ProviderRemote {
245	404	documents: Arc<Documents>,
246	405	path: String,
247		/// The image the last stamp had to read whole, which the step's read takes.
	406	/// The image the step read whole for a stamp, which its read takes.
248	407	read: Option<Vec<u8>>,
249	408	}
250	409
...	...	@@ -256,6 +415,82 @@ impl ProviderRemote {
256	415	read: None,
257	416	}
258	417	}
	418
	419	fn kept(&self) -> Vec<(Kept, std::time::SystemTime)> {
	420	let entries = crate::fs::read_dir(self.documents.published())
	421	.into_iter()
	422	.flatten();
	423	entries
	424	.filter_map(\|entry\| {
	425	let entry = entry.ok()?;
	426	let kept = Kept::named(entry.file_name().into_string().ok()?)?;
	427	Some((kept, entry.metadata().ok()?.modified().ok()?))
	428	})
	429	.collect()
	430	}
	431
	432	/// Keeps the image `transaction` published, in place of the one it was published on:
	433	/// that kept image with the transaction applied where this device kept it, else read back.
	434	fn keep(&self, written: Written<'_>, transaction: &Transaction) -> io::Result<()> {
	435	let folder = self.documents.published();
	436	let base = digest(transaction.base());
	437	let document = file_of(&transaction.base().header).ok_or(io::ErrorKind::InvalidData)?;
	438	let on = (self.kept().into_iter())
	439	.find(\|(kept, _)\| kept.document == hex(&document) && kept.image == base);
	440	let image = match (written, &on) {
	441	(Written::Image(image), _) => image.to_vec(),
	442	(Written::InPlace(_), Some((kept, _))) => {
	443	let mut image = crate::fs::read(folder.join(&kept.name))?;
	444	transaction
	445	.apply(&mut image)
	446	.map_err(\|error\| io::Error::new(io::ErrorKind::InvalidData, error))?;
	447	image
	448	}
	449	(Written::InPlace(file), None) => {
	450	file.seek(SeekFrom::Start(0))?;
	451	whole(file, crate::MAX_FILE_BYTES)?
	452	}
	453	};
	454	let name = format!(
	455	"{}-{}-{base}.one",
	456	hex(&document),
	457	digest(&transaction.stamp())
	458	);
	459	crate::fs::create_dir_all(&folder)?;
	460	let mut kept = tempfile::NamedTempFile::new_in(&folder)?;
	461	kept.write_all(&image)?;
	462	kept.as_file().sync_all()?;
	463	kept.persist(folder.join(name))
	464	.map_err(\|error\| error.error)?;
	465	if let Some((kept, _)) = on {
	466	crate::fs::remove_file(folder.join(kept.name))?;
	467	}
	468	Ok(())
	469	}
	470
	471	/// This device's own upload cut off, as a provider that writes non-atomically leaves it,
	472	/// put back and read again: the newest image kept that the torn bytes begin. Any other
	473	/// document that does not parse is left as it is.
	474	fn repair(&self, torn: &[u8]) -> io::Result<Vec<u8>> {
	475	let mut newest = None;
	476	for (kept, at) in self.kept() {
	477	let image = crate::fs::read(self.documents.published().join(&kept.name))?;
	478	if image.len() > torn.len() && image.starts_with(torn) {
	479	newest = newest.max(Some((at, image)));
	480	}
	481	}
	482	let (_, image) = newest.ok_or(io::ErrorKind::WouldBlock)?;
	483	let id = self.documents.id(&self.path)?;
	484	self.documents.writable(&id)?;
	485	self.documents
	486	.upload(&id, &image, \|_\| Ok(()))
	487	.map_err(\|error\| error.error)?;
	488	self.documents.read(&self.path, crate::MAX_FILE_BYTES)
	489	}
	490	}
	491
	492	fn parses(image: &[u8]) -> bool {
	493	Store::parse(image).is_ok_and(\|store\| store.checksum_mismatches.is_empty())
259	494	}
260	495
261	496	impl crate::Remote for ProviderRemote {
...	...	@@ -265,7 +500,7 @@ impl crate::Remote for ProviderRemote {
265	500	Some(image) => image,
266	501	None => self.documents.read(&self.path, crate::MAX_FILE_BYTES)?,
267	502	};
268		if !Store::parse(&image).is_ok_and(\|store\| store.checksum_mismatches.is_empty()) {
	503	if !parses(&image) {
269	504	return Err(io::Error::new(
270	505	io::ErrorKind::WouldBlock,
271	506	"The section was read mid-write",
...	...	@@ -275,6 +510,10 @@ impl crate::Remote for ProviderRemote {
275	510	}
276	511
277	512	fn stamp(&mut self) -> io::Result<Stamp> {
	513	if let Some(image) = &self.read {
	514	return Stamp::of(image)
	515	.map_err(\|error\| io::Error::new(io::ErrorKind::InvalidData, error));
	516	}
278	517	let (stamp, image) = self.documents.stamp(&self.path)?;
279	518	self.read = image;
280	519	Ok(stamp)
...	...	@@ -282,14 +521,63 @@ impl crate::Remote for ProviderRemote {
282	521
283	522	fn publish(&mut self, transaction: &Transaction) -> std::result::Result<(), CommitError> {
284	523	self.read = None;
285		self.documents
286		.write(&self.path, \|io\| transaction.commit(io))
	524	self.documents.write(
	525	&self.path,
	526	\|io\| transaction.commit(io),
	527	\|written\| self.keep(written, transaction),
	528	)
287	529	}
288	530
289	531	fn confirm(&mut self, base: &Stamp) -> std::result::Result<(), CommitError> {
290	532	self.read = None;
291	533	self.documents
292		.write(&self.path, \|io\| onestore::confirm(io, base))
	534	.write(&self.path, \|io\| onestore::confirm(io, base), \|_\| Ok(()))
	535	}
	536
	537	/// The images kept of this document other than the one it stands as; one whose upload
	538	/// the document still stands before is listed only once reads have had `LAG` to catch up.
	539	/// A document read torn is repaired first, once a step.
	540	fn versions(&mut self) -> io::Result<Vec<crate::Version>> {
	541	let (mut stamp, mut image) = self.documents.stamp(&self.path)?;
	542	if let Some(torn) = image.as_ref().filter(\|image\| !parses(image))
	543	&& let Ok(repaired) = self.repair(torn)
	544	&& let Ok(repaired_stamp) = Stamp::of(&repaired)
	545	{
	546	(stamp, image) = (repaired_stamp, Some(repaired));
	547	}
	548	self.read = image;
	549	// A header mid-write names no document; the step's read repairs it.
	550	let Some(document) = file_of(&stamp.header) else {
	551	return Ok(Vec::new());
	552	};
	553	let (document, current) = (hex(&document), digest(&stamp));
	554	let lagging = \|kept: &Kept, at: std::time::SystemTime\| {
	555	kept.base == current && at.elapsed().is_ok_and(\|elapsed\| elapsed < LAG)
	556	};
	557	Ok((self.kept().into_iter())
	558	.filter(\|(kept, at)\| {
	559	kept.document == document && kept.image != current && !lagging(kept, *at)
	560	})
	561	.map(\|(kept, _)\| crate::Version {
	562	id: kept.name,
	563	device: Some(self.documents.device.clone()),
	564	})
	565	.collect())
	566	}
	567
	568	fn version(&mut self, id: &str) -> io::Result<Vec<u8>> {
	569	crate::fs::read(self.documents.published().join(id))
	570	}
	571
	572	/// One that cannot merge, of another section under the document's name, is set aside
	573	/// whole, where `versions` lists it no more: no provider here takes a new document beside it.
	574	fn retire(&mut self, id: &str, keep: bool) -> io::Result<()> {
	575	let path = self.documents.published().join(id);
	576	if keep {
	577	crate::fs::rename(&path, path.with_extension("unmerged"))
	578	} else {
	579	crate::fs::remove_file(path)
	580	}
293	581	}
294	582	}
295	583
...	...	@@ -439,20 +727,24 @@ impl Storage for Tree {
439	727	}
440	728
441	729	fn place(&self, path: &str, ancestor: [u8; 16], name: &str) -> Result<()> {
442		Ok(self.documents.write(path, \|io\| {
	730	let place = \|io: &mut Io\| {
443	731	onestore::place(io, ancestor, name).map_err(\|error\| CommitError {
444	732	state: CommitState::Unknown,
445	733	error,
446	734	})
447		})?)
	735	};
	736	Ok(self.documents.write(path, place, \|_\| Ok(()))?)
448	737	}
449	738
450	739	fn commit(&self, path: &str, transaction: &Transaction) -> Result<()> {
451		Ok(self.documents.write(path, \|io\| transaction.commit(io))?)
	740	Ok(self
	741	.documents
	742	.write(path, \|io\| transaction.commit(io), \|_\| Ok(()))?)
452	743	}
453	744
454	745	fn confirm(&self, path: &str, base: &Stamp) -> std::result::Result<(), CommitError> {
455		self.documents.write(path, \|io\| onestore::confirm(io, base))
	746	self.documents
	747	.write(path, \|io\| onestore::confirm(io, base), \|_\| Ok(()))
456	748	}
457	749
458	750	fn supersede(&self, _: &str, _: &Stamp, _: &str) -> Result<()> {

crates/notebook/src/provider/tests.rs+403-81

...	...	@@ -12,9 +12,10 @@ use onestore::{
12	12	page::{Page, PageObject},
13	13	};
14	14	use std::{
	15	collections::HashMap,
15	16	fs,
16	17	ops::Range,
17		os::fd::OwnedFd,
	18	os::fd::{AsRawFd, OwnedFd},
18	19	path::Path,
19	20	thread,
20	21	time::{Duration, Instant},
...	...	@@ -25,8 +26,10 @@ const TOC: &str = "Open Notebook.onetoc2";
25	26	#[derive(Clone, Copy, PartialEq)]
26	27	enum Kind {
27	28	Seekable,
28		/// Pipes in every mode, `rw` included, with no metadata.
29		Stream,
	29	/// Pipes in every mode, `rw` included, with no metadata; `truncate` where `wt` is taken.
	30	Stream {
	31	truncate: bool,
	32	},
30	33	}
31	34
32	35	struct Node {
...	...	@@ -50,9 +53,19 @@ struct State {
50	53	/// The next `rw` open finds its document renamed to `Moved.one` under a new id, and a
51	54	/// document holding these bytes in its place.
52	55	swap: Option<Vec<u8>>,
	56	/// The next stream upload fails after this many bytes, leaving them as the document where
	57	/// `.1`, as a provider that writes non-atomically does, or else the document as it was.
	58	cut: Option<(u64, bool)>,
	59	/// Streams refuse to open for writing, as a provider out of space does.
	60	full: bool,
53	61	modes: Vec<String>,
	62	/// Each stream being uploaded, by descriptor: its document, whether `wt` truncates it, and
	63	/// what arrived with the cut that stopped it, if one did.
	64	uploads: HashMap<i32, (String, bool, Upload)>,
54	65	}
55	66
	67	type Upload = thread::JoinHandle<(Vec<u8>, Option<(u64, bool)>)>;
	68
56	69	struct Shared {
57	70	folder: tempfile::TempDir,
58	71	state: Mutex<State>,
...	...	@@ -86,7 +99,10 @@ impl Fake {
86	99	revoked: false,
87	100	delayed: false,
88	101	swap: None,
	102	cut: None,
	103	full: false,
89	104	modes: Vec::new(),
	105	uploads: HashMap::new(),
90	106	};
91	107	Self {
92	108	store: Arc::new(Shared {
...	...	@@ -105,7 +121,13 @@ impl Fake {
105	121	}
106	122
107	123	fn documents(&self, root: &str) -> Arc<Documents> {
108		Arc::new(Documents::new(self.clone(), root))
	124	let cache = self
	125	.store
	126	.folder
	127	.path()
	128	.join(format!("cache-{}", self.device));
	129	let device = format!("Device {}", self.device);
	130	Arc::new(Documents::new(self.clone(), root, cache, device))
109	131	}
110	132
111	133	fn state(&self) -> MutexGuard<'_, State> {
...	...	@@ -300,18 +322,59 @@ impl Provider for Fake {
300	322	.truncate(mode == "wt")
301	323	.open(file)
302	324	}
303		(Kind::Stream, "r") => {
	325	(Kind::Stream { .. }, "r") => {
304	326	let (reader, mut writer) = io::pipe()?;
305	327	let bytes = fs::read(self.file(id))?;
306	328	thread::spawn(move \|\| writer.write_all(&bytes));
307	329	Ok(File::from(OwnedFd::from(reader)))
308	330	}
309		(Kind::Stream, _) => Ok(File::from(OwnedFd::from(io::pipe()?.1))),
	331	(Kind::Stream { .. }, "rw") => Ok(File::from(OwnedFd::from(io::pipe()?.1))),
	332	// As Android's host reports a refused mode: unreachable, like most failures.
	333	(Kind::Stream { truncate: false }, "wt") => Err(io::Error::new(
	334	io::ErrorKind::NotConnected,
	335	"Unsupported mode",
	336	)),
	337	(Kind::Stream { .. }, _) if state.full => Err(io::Error::other("No space left")),
	338	(Kind::Stream { .. }, _) => {
	339	let (reader, writer) = io::pipe()?;
	340	let cut = state.cut.take();
	341	let upload = thread::spawn(move \|\| {
	342	let mut received = Vec::new();
	343	let at = cut.map_or(u64::MAX, \|(at, _)\| at);
	344	let _ = reader.take(at).read_to_end(&mut received);
	345	(received, cut)
	346	});
	347	let writer = File::from(OwnedFd::from(writer));
	348	let upload = (id.to_owned(), mode == "wt", upload);
	349	state.uploads.insert(writer.as_raw_fd(), upload);
	350	Ok(writer)
	351	}
310	352	}
311	353	}
312	354
313		fn close(&self, _: File) -> io::Result<()> {
314		Ok(())
	355	fn close(&self, file: File) -> io::Result<()> {
	356	let upload = self.state().uploads.remove(&file.as_raw_fd());
	357	drop(file);
	358	let Some((id, truncate, upload)) = upload else {
	359	return Ok(());
	360	};
	361	let (received, cut) = upload.join().unwrap();
	362	if let Some((_, torn)) = cut {
	363	if torn {
	364	fs::write(self.file(&id), received)?;
	365	}
	366	return Err(io::Error::other("The upload failed"));
	367	}
	368	let mut bytes = if truncate {
	369	Vec::new()
	370	} else {
	371	fs::read(self.file(&id))?
	372	};
	373	if bytes.len() < received.len() {
	374	bytes.resize(received.len(), 0);
	375	}
	376	bytes[..received.len()].copy_from_slice(&received);
	377	fs::write(self.file(&id), bytes)
315	378	}
316	379
317	380	fn rename(&self, id: &str, name: &str) -> io::Result<String> {
...	...	@@ -326,10 +389,33 @@ impl Provider for Fake {
326	389	}
327	390	}
328	391
	392	/// A section holding `text`, edited once since it was made, as a section OneNote has written
	393	/// is: the first edit of a page `create_section` makes starts a fresh revision chain, so its
	394	/// pages' later states trace back to that edit, not to the section as made.
	395	fn section_image(name: &str, text: &str) -> Vec<u8> {
	396	let (first, rest) = text.split_at(1);
	397	let mut image = onestore::create_section(name, rest, "Author").unwrap();
	398	let arena = onestore::Arena::default();
	399	let mut section = onestore::Section::open(&arena, image.clone()).unwrap();
	400	let space = section.pages().unwrap()[0].0;
	401	let op = PageOp::Text {
	402	text: paragraph(&section.page(space).unwrap()).0,
	403	range: 0..0,
	404	with: first.into(),
	405	};
	406	let edit = Edit {
	407	at: crate::now(),
	408	ops: vec![Op::Page { space, op }],
	409	};
	410	section.apply("Author", &edit).unwrap();
	411	section.seal().unwrap().unwrap().apply(&mut image).unwrap();
	412	image
	413	}
	414
329	415	/// A notebook folder: its TOC and two sections it lists, placed under it.
330	416	fn notebook(fake: &Fake) {
331		let mut notes = onestore::create_section("Notes.one", "Original text", "Author").unwrap();
332		let mut ideas = onestore::create_section("Ideas.one", "First idea", "Author").unwrap();
	417	let mut notes = section_image("Notes.one", "Original text");
	418	let mut ideas = section_image("Ideas.one", "First idea");
333	419	let file = \|image: &[u8]\| Store::parse(image).unwrap().header.file_id;
334	420	let toc = onestore::create_table_of_contents(
335	421	TOC,
...	...	@@ -432,9 +518,8 @@ fn status(section: &session::Section, accept: impl Fn(&SyncStatus) -> bool) -> S
432	518	fn seekable_documents_take_commits_in_place_and_rebase_on_another_devices() {
433	519	let fake = Fake::new(Kind::Seekable);
434	520	notebook(&fake);
435		let cache = tempfile::tempdir().unwrap();
436	521	let documents = fake.documents("root");
437		let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
	522	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
438	523	assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]);
439	524	let section = open(&notebook, &documents, "Notes.one");
440	525	let before = fake.bytes("Notes.one").len();
...	...	@@ -446,8 +531,7 @@ fn seekable_documents_take_commits_in_place_and_rebase_on_another_devices() {
446	531	assert!(fake.modes().iter().all(\|mode\| mode == "r" \|\| mode == "rw"));
447	532
448	533	let elsewhere = fake.device(1).documents("root");
449		let other_cache = tempfile::tempdir().unwrap();
450		let other = Notebook::open_provider(Arc::clone(&elsewhere), other_cache.path()).unwrap();
	534	let other = Notebook::open_provider(Arc::clone(&elsewhere)).unwrap();
451	535	let other = open(&other, &elsewhere, "Notes.one");
452	536	let id = typed(&other, 0..0, "World ");
453	537	published(&other, id);
...	...	@@ -467,44 +551,24 @@ fn seekable_documents_take_commits_in_place_and_rebase_on_another_devices() {
467	551	}
468	552
469	553	#[test]
470		fn stream_only_documents_are_read_and_never_opened_to_write() {
471		let fake = Fake::new(Kind::Stream);
472		notebook(&fake);
473		let cache = tempfile::tempdir().unwrap();
474		let documents = fake.documents("root");
475		let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
476		let section = open(&notebook, &documents, "Notes.one");
477		let id = typed(&section, 13..13, " mine");
478		let waiting = status(&section, \|status\| status.error.is_some());
479		assert_eq!(waiting.state(), SyncState::ReadOnly);
480		assert_eq!(waiting.queued, 1);
481		assert!(fake.modes().iter().all(\|mode\| mode == "r"));
482
483		// Another device, which can write in place, changes the document.
484		let elsewhere = Fake::new(Kind::Seekable);
485		for name in [TOC, "Notes.one", "Ideas.one"] {
486		elsewhere.add(name, &fake.bytes(name));
	554	fn stream_only_documents_take_whole_images_and_never_open_to_write_in_place() {
	555	for truncate in [true, false] {
	556	let fake = Fake::new(Kind::Stream { truncate });
	557	notebook(&fake);
	558	let documents = fake.documents("root");
	559	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
	560	let section = open(&notebook, &documents, "Ideas.one");
	561	let id = typed(&section, 0..0, "Second. ");
	562	published(&section, id);
	563	let id = typed(&section, 0..0, "Third. ");
	564	published(&section, id);
	565	assert_eq!(stored(fake.bytes("Ideas.one")), "Third. Second. First idea");
	566	let modes = fake.modes();
	567	let count = \|wanted: &str\| modes.iter().filter(\|mode\| *mode == wanted).count();
	568	assert_eq!(count("rw"), 0, "{modes:?}");
	569	assert_eq!((count("wt"), count("w")), (2, if truncate { 0 } else { 2 }));
	570	section.close().unwrap();
487	571	}
488		let other_documents = elsewhere.documents("root");
489		let other_cache = tempfile::tempdir().unwrap();
490		let other = Notebook::open_provider(Arc::clone(&other_documents), other_cache.path()).unwrap();
491		let other = open(&other, &other_documents, "Notes.one");
492		let theirs = typed(&other, 0..0, "Theirs ");
493		published(&other, theirs);
494		other.close().unwrap();
495		fs::write(
496		fake.file(&fake.id("Notes.one")),
497		elsewhere.bytes("Notes.one"),
498		)
499		.unwrap();
500		until("the change was read", \|\| {
501		section.wake();
502		text(&section) == "Theirs Original text mine"
503		});
504		assert_eq!(section.pending().unwrap()[0].id, id);
505		assert_eq!(stored(fake.bytes("Notes.one")), "Theirs Original text");
506		assert!(fake.modes().iter().all(\|mode\| mode == "r"));
507		section.close().unwrap();
508	572	}
509	573
510	574	#[test]
...	...	@@ -512,9 +576,8 @@ fn documents_without_a_write_grant_are_only_read() {
512	576	let fake = Fake::new(Kind::Seekable);
513	577	notebook(&fake);
514	578	fake.set(\|state\| state.writable = false);
515		let cache = tempfile::tempdir().unwrap();
516	579	let documents = fake.documents("root");
517		let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
	580	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
518	581	let section = open(&notebook, &documents, "Notes.one");
519	582	let before = fake.bytes("Notes.one");
520	583	typed(&section, 0..0, "Mine ");
...	...	@@ -536,9 +599,8 @@ fn a_write_another_device_sees_late_rebases_it_once_visible() {
536	599	notebook(&fake);
537	600	fake.set(\|state\| state.delayed = true);
538	601	let (writer, reader) = (fake.documents("root"), fake.device(1).documents("root"));
539		let caches = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
540		let writing = Notebook::open_provider(Arc::clone(&writer), caches.0.path()).unwrap();
541		let reading = Notebook::open_provider(Arc::clone(&reader), caches.1.path()).unwrap();
	602	let writing = Notebook::open_provider(Arc::clone(&writer)).unwrap();
	603	let reading = Notebook::open_provider(Arc::clone(&reader)).unwrap();
542	604	let (writing, reading) = (
543	605	open(&writing, &writer, "Notes.one"),
544	606	open(&reading, &reader, "Notes.one"),
...	...	@@ -567,9 +629,8 @@ fn a_write_another_device_sees_late_rebases_it_once_visible() {
567	629	fn a_cold_start_while_offline_opens_the_last_listing_and_queued_edits() {
568	630	let fake = Fake::new(Kind::Seekable);
569	631	notebook(&fake);
570		let cache = tempfile::tempdir().unwrap();
571	632	let documents = fake.documents("root");
572		let mut notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
	633	let mut notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
573	634	let background = Background::provider(Arc::clone(&documents), \|\| {}).unwrap();
574	635	background.watch(notebook.replicas());
575	636	until("every section has its offline copy", \|\| {
...	...	@@ -589,7 +650,7 @@ fn a_cold_start_while_offline_opens_the_last_listing_and_queued_edits() {
589	650
590	651	// A new process: the provider fails throughout.
591	652	let documents = fake.documents("root");
592		let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
	653	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
593	654	assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]);
594	655	let notes = open(&notebook, &documents, "Notes.one");
595	656	assert_eq!(text(&notes), "Original text");
...	...	@@ -608,9 +669,8 @@ fn a_cold_start_while_offline_opens_the_last_listing_and_queued_edits() {
608	669	fn a_revoked_grant_is_not_offline_and_keeps_the_replicas() {
609	670	let fake = Fake::new(Kind::Seekable);
610	671	notebook(&fake);
611		let cache = tempfile::tempdir().unwrap();
612	672	let documents = fake.documents("root");
613		let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
	673	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
614	674	let replica = notebook.replica_path("Notes.one").unwrap();
615	675	let section = open(&notebook, &documents, "Notes.one");
616	676	fake.set(\|state\| state.revoked = true);
...	...	@@ -622,7 +682,7 @@ fn a_revoked_grant_is_not_offline_and_keeps_the_replicas() {
622	682	);
623	683	section.close().unwrap();
624	684	drop(notebook);
625		let refused = Notebook::open_provider(fake.documents("root"), cache.path()).err();
	685	let refused = Notebook::open_provider(fake.documents("root")).err();
626	686	assert!(
627	687	matches!(
628	688	&refused,
...	...	@@ -638,7 +698,7 @@ fn a_revoked_grant_is_not_offline_and_keeps_the_replicas() {
638	698
639	699	fake.set(\|state\| state.revoked = false);
640	700	let documents = fake.documents("root");
641		let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
	701	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
642	702	let section = open(&notebook, &documents, "Notes.one");
643	703	published(&section, id);
644	704	assert_eq!(stored(fake.bytes("Notes.one")), "Kept Original text");
...	...	@@ -649,9 +709,8 @@ fn a_revoked_grant_is_not_offline_and_keeps_the_replicas() {
649	709	fn documents_renamed_to_new_ids_are_listed_again() {
650	710	let fake = Fake::new(Kind::Seekable);
651	711	notebook(&fake);
652		let cache = tempfile::tempdir().unwrap();
653	712	let documents = fake.documents("root");
654		let mut notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
	713	let mut notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
655	714	let section = open(&notebook, &documents, "Notes.one");
656	715	let id = typed(&section, 0..0, "One ");
657	716	published(&section, id);
...	...	@@ -691,9 +750,8 @@ fn of_documents_sharing_a_name_one_lists_and_the_rest_are_unavailable() {
691	750	"Notes.one",
692	751	&onestore::create_section("Notes.one", "A twin", "Author").unwrap(),
693	752	);
694		let cache = tempfile::tempdir().unwrap();
695	753	let documents = fake.documents("root");
696		let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
	754	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
697	755	assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]);
698	756	let unavailable = &notebook.catalog().unavailable;
699	757	assert_eq!(unavailable.len(), 1);
...	...	@@ -707,9 +765,8 @@ fn of_documents_sharing_a_name_one_lists_and_the_rest_are_unavailable() {
707	765	fn a_document_taking_the_name_between_check_and_write_is_not_written() {
708	766	let fake = Fake::new(Kind::Seekable);
709	767	notebook(&fake);
710		let cache = tempfile::tempdir().unwrap();
711	768	let documents = fake.documents("root");
712		let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
	769	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
713	770	let section = open(&notebook, &documents, "Notes.one");
714	771	let id = typed(&section, 0..0, "Checked ");
715	772	published(&section, id);
...	...	@@ -761,9 +818,8 @@ fn conflicting_edits_leave_a_conflict_page_in_the_document() {
761	818	let fake = Fake::new(Kind::Seekable);
762	819	notebook(&fake);
763	820	let (mine, theirs) = (fake.documents("root"), fake.device(1).documents("root"));
764		let caches = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
765		let local = Notebook::open_provider(Arc::clone(&mine), caches.0.path()).unwrap();
766		let remote = Notebook::open_provider(Arc::clone(&theirs), caches.1.path()).unwrap();
	821	let local = Notebook::open_provider(Arc::clone(&mine)).unwrap();
	822	let remote = Notebook::open_provider(Arc::clone(&theirs)).unwrap();
767	823	let (local, remote) = (
768	824	open(&local, &mine, "Notes.one"),
769	825	open(&remote, &theirs, "Notes.one"),
...	...	@@ -793,10 +849,8 @@ fn a_lone_document_opens_offline_from_its_replica() {
793	849	let fake = Fake::new(Kind::Seekable);
794	850	let image = onestore::create_section("Lone.one", "Alone", "Author").unwrap();
795	851	let lone = fake.add("Lone.one", &image);
796		let cache = tempfile::tempdir().unwrap();
797		let open = \|fake: &Fake\| {
798		session::Section::open_provider(fake.documents(&lone), cache.path(), false, \|\| {}).unwrap()
799		};
	852	let open =
	853	\|fake: &Fake\| session::Section::open_provider(fake.documents(&lone), false, \|\| {}).unwrap();
800	854	let section = open(&fake);
801	855	let id = typed(&section, 0..0, "Not ");
802	856	published(&section, id);
...	...	@@ -817,15 +871,13 @@ fn a_lone_document_opens_offline_from_its_replica() {
817	871	fn a_provider_that_lists_but_cannot_open_opens_sections_from_their_replicas() {
818	872	let fake = Fake::new(Kind::Seekable);
819	873	notebook(&fake);
820		let cache = tempfile::tempdir().unwrap();
821	874	let documents = fake.documents("root");
822		let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
	875	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
823	876	open(&notebook, &documents, "Notes.one").close().unwrap();
824	877	drop(notebook);
825	878	// Another device changes the section, so it lists otherwise than when last read.
826	879	let elsewhere = fake.device(1).documents("root");
827		let other_cache = tempfile::tempdir().unwrap();
828		let other = Notebook::open_provider(Arc::clone(&elsewhere), other_cache.path()).unwrap();
	880	let other = Notebook::open_provider(Arc::clone(&elsewhere)).unwrap();
829	881	let other = open(&other, &elsewhere, "Notes.one");
830	882	let id = typed(&other, 0..0, "Theirs ");
831	883	published(&other, id);
...	...	@@ -833,7 +885,7 @@ fn a_provider_that_lists_but_cannot_open_opens_sections_from_their_replicas() {
833	885	fake.set(\|state\| state.unopenable = true);
834	886
835	887	let documents = fake.documents("root");
836		let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
	888	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
837	889	assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]);
838	890	let section = open(&notebook, &documents, "Notes.one");
839	891	assert_eq!(text(&section), "Original text");
...	...	@@ -846,3 +898,273 @@ fn a_provider_that_lists_but_cannot_open_opens_sections_from_their_replicas() {
846	898	assert_eq!(stored(fake.bytes("Notes.one")), "Theirs Original text mine");
847	899	section.close().unwrap();
848	900	}
	901
	902	/// The images a device keeps of what it published, for every notebook in its cache.
	903	fn kept_images(fake: &Fake) -> Vec<String> {
	904	let cache = fake
	905	.store
	906	.folder
	907	.path()
	908	.join(format!("cache-{}", fake.device));
	909	let folders = fs::read_dir(cache.join("replicas")).into_iter().flatten();
	910	folders
	911	.filter_map(\|folder\| fs::read_dir(folder.ok()?.path().join("published")).ok())
	912	.flatten()
	913	.map(\|entry\| entry.unwrap().file_name().to_string_lossy().into_owned())
	914	.filter(\|name\| name.ends_with(".one"))
	915	.collect()
	916	}
	917
	918	/// The images a device keeps that Notes.one does not stand as, still to merge.
	919	fn unmerged(fake: &Fake) -> Vec<String> {
	920	let current = digest(&Stamp::of(&fake.bytes("Notes.one")).unwrap());
	921	let mut images = kept_images(fake);
	922	images.retain(\|name\| Kept::named(name.clone()).is_some_and(\|kept\| kept.image != current));
	923	images
	924	}
	925
	926	#[test]
	927	fn a_publication_another_devices_upload_replaced_merges_back() {
	928	let fake = Fake::new(Kind::Seekable);
	929	notebook(&fake);
	930	fake.set(\|state\| state.delayed = true);
	931	let (mine, theirs) = (fake.documents("root"), fake.device(1).documents("root"));
	932	let (local, remote) = (
	933	Notebook::open_provider(Arc::clone(&mine)).unwrap(),
	934	Notebook::open_provider(Arc::clone(&theirs)).unwrap(),
	935	);
	936	let (local, remote) = (
	937	open(&local, &mine, "Notes.one"),
	938	open(&remote, &theirs, "Notes.one"),
	939	);
	940	let id = typed(&local, 0..0, "Alpha ");
	941	published(&local, id);
	942	let id = typed(&remote, 13..13, " beta");
	943	published(&remote, id);
	944	// Both uploads land; the later one, which never saw the first, is the document.
	945	fake.publish_staged();
	946	assert_eq!(stored(fake.bytes("Notes.one")), "Original text beta");
	947	until("the lost publication merged back", \|\| {
	948	local.wake();
	949	fake.publish_staged();
	950	stored(fake.bytes("Notes.one")) == "Alpha Original text beta"
	951	});
	952	until("both devices show it", \|\| {
	953	local.wake();
	954	remote.wake();
	955	text(&local) == "Alpha Original text beta" && text(&remote) == text(&local)
	956	});
	957	assert!(local.conflicts().unwrap().is_empty());
	958	assert!(unmerged(&fake).is_empty());
	959	local.close().unwrap();
	960	remote.close().unwrap();
	961	}
	962
	963	/// Ages the images a device keeps past `LAG`, as if reads had long had time to catch up.
	964	fn aged(fake: &Fake) {
	965	let cache = fake
	966	.store
	967	.folder
	968	.path()
	969	.join(format!("cache-{}", fake.device));
	970	for folder in fs::read_dir(cache.join("replicas")).unwrap() {
	971	for image in fs::read_dir(folder.unwrap().path().join("published"))
	972	.into_iter()
	973	.flatten()
	974	{
	975	let file = File::options()
	976	.write(true)
	977	.open(image.unwrap().path())
	978	.unwrap();
	979	let past = std::time::SystemTime::now() - LAG - Duration::from_secs(1);
	980	file.set_modified(past).unwrap();
	981	}
	982	}
	983	}
	984
	985	fn failed_attempt(section: &session::Section) {
	986	until("the upload failed", \|\| {
	987	section.wake();
	988	(section.events().iter()).any(\|event\| matches!(event, Event::Unreachable(_)))
	989	});
	990	}
	991
	992	#[test]
	993	fn an_upload_the_provider_dropped_merges_back_once_reads_could_have_caught_up() {
	994	let fake = Fake::new(Kind::Stream { truncate: true });
	995	notebook(&fake);
	996	let documents = fake.documents("root");
	997	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
	998	let section = open(&notebook, &documents, "Notes.one");
	999	fake.set(\|state\| state.cut = Some((2000, false)));
	1000	let id = typed(&section, 0..0, "Cut ");
	1001	failed_attempt(&section);
	1002	// The document stands as the upload was made on: reads may lag it, so it waits.
	1003	for _ in 0..5 {
	1004	section.wake();
	1005	thread::sleep(Duration::from_millis(50));
	1006	}
	1007	assert_eq!(kept_images(&fake).len(), 1);
	1008	assert_eq!(stored(fake.bytes("Notes.one")), "Original text");
	1009	aged(&fake);
	1010	published(&section, id);
	1011	assert_eq!(stored(fake.bytes("Notes.one")), "Cut Original text");
	1012	section.close().unwrap();
	1013	}
	1014
	1015	#[test]
	1016	fn a_torn_upload_on_a_provider_writing_in_place_is_put_back() {
	1017	for at in [2000, 500] {
	1018	let fake = Fake::new(Kind::Stream { truncate: true });
	1019	notebook(&fake);
	1020	let documents = fake.documents("root");
	1021	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
	1022	let section = open(&notebook, &documents, "Notes.one");
	1023	fake.set(\|state\| state.cut = Some((at, true)));
	1024	let id = typed(&section, 0..0, "Torn ");
	1025	published(&section, id);
	1026	assert_eq!(stored(fake.bytes("Notes.one")), "Torn Original text");
	1027	until("the kept image was retired", \|\| {
	1028	section.wake();
	1029	unmerged(&fake).is_empty()
	1030	});
	1031	section.close().unwrap();
	1032	}
	1033	}
	1034
	1035	#[test]
	1036	fn an_upload_refused_before_it_began_is_retried_not_merged() {
	1037	let fake = Fake::new(Kind::Stream { truncate: true });
	1038	notebook(&fake);
	1039	let documents = fake.documents("root");
	1040	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
	1041	let section = open(&notebook, &documents, "Notes.one");
	1042	fake.set(\|state\| state.full = true);
	1043	let id = typed(&section, 0..0, "Once ");
	1044	failed_attempt(&section);
	1045	assert!(kept_images(&fake).is_empty());
	1046	assert_eq!(section.pending().unwrap().len(), 1);
	1047	fake.set(\|state\| state.full = false);
	1048	published(&section, id);
	1049	assert_eq!(stored(fake.bytes("Notes.one")), "Once Original text");
	1050	section.close().unwrap();
	1051	}
	1052
	1053	#[test]
	1054	fn devices_whose_uploads_replace_each_other_converge_with_every_string_once() {
	1055	for seed in 1..=4u64 {
	1056	let fake = Fake::new(Kind::Seekable);
	1057	notebook(&fake);
	1058	fake.set(\|state\| state.delayed = true);
	1059	let devices: Vec<_> = (0..3)
	1060	.map(\|device\| {
	1061	let documents = fake.device(device).documents("root");
	1062	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
	1063	open(&notebook, &documents, "Notes.one")
	1064	})
	1065	.collect();
	1066	let mut random = seed;
	1067	let mut next = \|bound: u64\| {
	1068	random = random
	1069	.wrapping_mul(6364136223846793005)
	1070	.wrapping_add(1442695040888963407);
	1071	(random >> 33) % bound
	1072	};
	1073	let mut strings = Vec::new();
	1074	for round in 0..6 {
	1075	let device = next(3) as usize;
	1076	let string = format!("s{seed}r{round}d{device}");
	1077	typed(&devices[device], 0..0, &format!("{string} "));
	1078	strings.push(string);
	1079	devices[device].wake();
	1080	thread::sleep(Duration::from_millis(next(150)));
	1081	if next(2) == 0 {
	1082	fake.publish_staged();
	1083	}
	1084	}
	1085	until("every device converged", \|\| {
	1086	for device in &devices {
	1087	device.wake();
	1088	}
	1089	thread::sleep(Duration::from_millis(100));
	1090	fake.publish_staged();
	1091	let file = stored(fake.bytes("Notes.one"));
	1092	devices
	1093	.iter()
	1094	.all(\|device\| device.pending().unwrap().is_empty() && text(device) == file)
	1095	&& (0..3).all(\|device\| unmerged(&fake.device(device)).is_empty())
	1096	});
	1097	let arena = onestore::Arena::default();
	1098	let mut section = onestore::Section::open(&arena, fake.bytes("Notes.one")).unwrap();
	1099	let mut conflicts = String::new();
	1100	for (_, pages) in section.conflicts().unwrap() {
	1101	for conflict in pages {
	1102	conflicts += &paragraph(&section.page(conflict.space).unwrap()).1;
	1103	}
	1104	}
	1105	let file = stored(fake.bytes("Notes.one"));
	1106	for string in &strings {
	1107	let shown = file.matches(string.as_str()).count();
	1108	assert!(
	1109	shown <= 1,
	1110	"seed {seed}: {string} shows {shown} times in {file}"
	1111	);
	1112	assert!(
	1113	shown == 1 \|\| conflicts.contains(string.as_str()),
	1114	"seed {seed}: {string} is gone from {file}"
	1115	);
	1116	}
	1117	for device in devices {
	1118	device.close().unwrap();
	1119	}
	1120	}
	1121	}
	1122
	1123	#[test]
	1124	fn a_document_this_device_did_not_write_that_does_not_parse_is_never_rewritten() {
	1125	let fake = Fake::new(Kind::Stream { truncate: true });
	1126	notebook(&fake);
	1127	let documents = fake.documents("root");
	1128	let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap();
	1129	let section = open(&notebook, &documents, "Notes.one");
	1130	let id = typed(&section, 0..0, "Mine ");
	1131	published(&section, id);
	1132	// The same document as another writer took it on, cut off, then with checksums that fail.
	1133	let mut theirs = fake.bytes("Notes.one");
	1134	let arena = onestore::Arena::default();
	1135	let mut section_there = onestore::Section::open(&arena, theirs.clone()).unwrap();
	1136	let space = section_there.pages().unwrap()[0].0;
	1137	let op = PageOp::Text {
	1138	text: paragraph(&section_there.page(space).unwrap()).0,
	1139	range: 0..0,
	1140	with: "Theirs ".into(),
	1141	};
	1142	let edit = Edit {
	1143	at: crate::now(),
	1144	ops: vec![Op::Page { space, op }],
	1145	};
	1146	section_there.apply("Them", &edit).unwrap();
	1147	let sealed = section_there.seal().unwrap().unwrap();
	1148	sealed.apply(&mut theirs).unwrap();
	1149	let mut damaged = theirs.clone();
	1150	let at = (1024..damaged.len())
	1151	.find(\|at\| {
	1152	let mut flipped = damaged.clone();
	1153	flipped[*at] ^= 0xff;
	1154	Store::parse(&flipped).is_ok_and(\|store\| !store.checksum_mismatches.is_empty())
	1155	})
	1156	.unwrap();
	1157	damaged[at] ^= 0xff;
	1158	for unreadable in [theirs[..theirs.len() - 100].to_vec(), damaged] {
	1159	fs::write(fake.file(&fake.id("Notes.one")), &unreadable).unwrap();
	1160	let since = fake.modes().len();
	1161	for _ in 0..5 {
	1162	section.wake();
	1163	thread::sleep(Duration::from_millis(50));
	1164	}
	1165	status(&section, \|status\| status.state() == SyncState::InUse);
	1166	assert_eq!(fake.bytes("Notes.one"), unreadable);
	1167	assert!(fake.modes()[since..].iter().all(\|mode\| mode == "r"));
	1168	}
	1169	section.close().unwrap();
	1170	}

crates/notebook/src/session.rs+5-7

...	...	@@ -517,11 +517,9 @@ impl Notebook {
517	517	/// Opens the notebook whose folder is the root of `documents`. Sections open through
518	518	/// `Section::resume_provider` with the catalog's paths; while the provider can't be
519	519	/// reached, the notebook opens as its folders were last listed.
520		pub fn open_provider(
521		documents: Arc<crate::provider::Documents>,
522		cache: impl AsRef<Path>,
523		) -> Result<Self> {
524		let listed = listing(cache.as_ref(), &documents.location()).with_extension("entries.json");
	520	pub fn open_provider(documents: Arc<crate::provider::Documents>) -> Result<Self> {
	521	let cache = documents.cache.clone();
	522	let listed = listing(&cache, &documents.location()).with_extension("entries.json");
525	523	Self::with(
526	524	Box::new(crate::provider::Tree { documents, listed }),
527	525	None,
...	...	@@ -2418,11 +2416,11 @@ impl Section {
2418	2416	/// its location has.
2419	2417	pub fn open_provider(
2420	2418	documents: Arc<crate::provider::Documents>,
2421		cache: impl AsRef<Path>,
2422	2419	offline: bool,
2423	2420	notify: impl Fn() + Send + Sync + 'static,
2424	2421	) -> Result<Self> {
2425		let (cache, location) = (cache.as_ref(), documents.location());
	2422	let (cache, location) = (documents.cache.clone(), documents.location());
	2423	let cache = cache.as_path();
2426	2424	let file = PathBuf::from(&documents.root);
2427	2425	let remote = move \|\| Ok(crate::provider::ProviderRemote::new(&documents, ""));
2428	2426	let notify = Arc::new(notify);

crates/notebook/src/sync.rs+1-1

...	...	@@ -259,7 +259,7 @@ impl Replica {
259	259	awaited(remote, \|remote\| remote.publish(&transaction)).await?;
260	260	false
261	261	}
262		// A version this cannot merge is kept whole rather than lost.
	262	// A version this cannot merge is retired to be kept whole, never lost.
263	263	Ok(Merged::Foreign) \| Err(Error::Document(_) \| Error::Rejected(_)) => true,
264	264	Err(error) => return Err(error),
265	265	};