diff --git a/apps/android/jni/snowbound.c b/apps/android/jni/snowbound.c index 44608bf65debe41dfe154883eceef90ea50ad3fa..61f7ff4a42c51007126216daa04d0836c85646b6 100644 --- a/apps/android/jni/snowbound.c +++ b/apps/android/jni/snowbound.c @@ -247,12 +247,13 @@ JNIEXPORT void JNICALL PROVIDER(renamed)(JNIEnv *env, jclass cls, jlong renamed, } JNIEXPORT jlong JNICALL JNI(provided)(JNIEnv *env, jclass cls, jstring root, jboolean folder, jboolean local, - jstring cache) { + jstring device, jstring cache) { (void)cls; - char *r = utf8(env, root), *c = r ? utf8(env, cache) : NULL, *error = NULL; - const Library *library = c ? sb_library_provider(r, folder, local, c, &error) : NULL; + char *r = utf8(env, root), *d = r ? utf8(env, device) : NULL, *c = d ? utf8(env, cache) : NULL; + char *error = NULL; + const Library *library = c ? sb_library_provider(r, folder, local, d, c, &error) : NULL; if (c && !library) fail(env, error, "The notebook can't be opened"); - free(r), free(c); + free(r), free(d), free(c); return (jlong)(intptr_t)library; } diff --git a/apps/android/src/net/paperclover/snowbound/Editor.java b/apps/android/src/net/paperclover/snowbound/Editor.java index e42b903e12f19e985e6ee9bef3afaba520c2f26d..9f86c671266b262071965913b5b4694cb4454f5a 100644 --- a/apps/android/src/net/paperclover/snowbound/Editor.java +++ b/apps/android/src/net/paperclover/snowbound/Editor.java @@ -1,9 +1,11 @@ package net.paperclover.snowbound; import android.content.Context; +import android.os.Build; import android.os.Handler; import android.os.Looper; import android.util.Log; +import android.provider.Settings; import android.view.Surface; import java.io.File; import java.text.DateFormat; @@ -62,7 +64,8 @@ final class Editor { if (picked != null && granted) { Provider.writable = picked.writable(); try { - opened = provided(picked.root(), picked.folder, picked.local(), cache.getPath()); + opened = provided(picked.root(), picked.folder, picked.local(), device(context), + cache.getPath()); path = firstSection(opened); first = section(opened, path); } catch (IllegalStateException | JSONException error) { @@ -139,6 +142,12 @@ final class Editor { return false; } + /** The name the user gave this device, as conflict pages name who made them. */ + private static String device(Context context) { + String name = Settings.Global.getString(context.getContentResolver(), Settings.Global.DEVICE_NAME); + return name != null ? name : Build.MODEL; + } + /** Called on the section's thread when it has events. */ static void woken() { MAIN.post(Editor::poll); @@ -232,6 +241,6 @@ final class Editor { static native void reload(long view, boolean discard); static native void freeView(long view); static native boolean close(long library, long section, double seconds); - static native long provided(String root, boolean folder, boolean local, String cache); + static native long provided(String root, boolean folder, boolean local, String device, String cache); static native String syncStatus(long library); } diff --git a/crates/mobile/include/snowbound.h b/crates/mobile/include/snowbound.h index 123aed94ec3579195971a8a0f72916323cc4f0d3..4be91f7377fb40352e78b4016ab56cc5217d42f4 100644 --- a/crates/mobile/include/snowbound.h +++ b/crates/mobile/include/snowbound.h @@ -39,7 +39,7 @@ typedef void (*sb_document_found)(void *context, const char *id, const char *nam typedef struct sb_provider { int32_t (*document)(const char *id, sb_document_found found, void *context); int32_t (*children)(const char *id, sb_document_found found, void *context); - // mode is "r" or "rw"; fd is a duplicate Snowbound owns and closes, handle the descriptor + // mode is "r", "rw", "wt" or "w"; fd is a duplicate Snowbound owns and closes, handle the descriptor // the host keeps open: detachFd would end the channel checkError reads. A nonzero status // sets neither; a negative fd with status 0 has its handle closed. int32_t (*open)(const char *id, const char *mode, int32_t *fd, uint64_t *handle); @@ -50,7 +50,8 @@ typedef struct sb_provider { void sb_set_provider(const sb_provider *provider); const Library *sb_library_open(const char *path, const char *cache, bool local, char **error); -const Library *sb_library_provider(const char *root, bool folder, bool local, const char *cache, char **error); +const Library *sb_library_provider(const char *root, bool folder, bool local, const char *device, const char *cache, + char **error); const Library *sb_library_server(const char *address, const char *share, const char *user, const char *password, const char *domain, const char *root, const char *cache, char **error); // Callback/token live until the blocking join returns; false cancels, approving reports host approval. diff --git a/crates/mobile/src/library.rs b/crates/mobile/src/library.rs index 9e4f19042262c61b0bc85861d0bccb0aa206553d..3bd33c8eee027506d6174694ee2a8c3b615635af 100644 --- a/crates/mobile/src/library.rs +++ b/crates/mobile/src/library.rs @@ -512,7 +512,7 @@ impl Library { /// registered provider keeps (`sb_set_provider`). A provider `local` to this device /// publishes each edit at once; any other waits for a pause in typing. While the provider /// can't be reached, the folder opens as last listed and sections from their offline copies. - fn provider(root: &str, folder: bool, local: bool, cache: &Path) -> Result { + fn provider(root: &str, folder: bool, local: bool, device: &str, cache: &Path) -> Result { let host = crate::provider::Host::default(); let name = (!folder) .then(|| { @@ -523,9 +523,9 @@ impl Library { ) }) .flatten(); - let documents = Arc::new(Documents::new(host, root)); + let documents = Arc::new(Documents::new(host, root, cache, device)); let (notebook, place, background) = if folder { - let mut notebook = Notebook::open_provider(Arc::clone(&documents), cache)?; + let mut notebook = Notebook::open_provider(Arc::clone(&documents))?; let background = Background::provider(Arc::clone(&documents), sync_woken)?; background.watch(notebook.replicas()); (Some(notebook), Place::Provider(documents), Some(background)) @@ -824,7 +824,6 @@ impl Library { )?), Place::Document(documents, _) => Ok(session::Section::open_provider( Arc::clone(documents), - &self.cache, offline, notify, )?), @@ -2224,22 +2223,32 @@ pub unsafe extern "C" fn sb_library_flush( /// The notebook folder whose document is `root`, or with `folder` false the lone section /// `root` names, as the provider `sb_set_provider` registered keeps it; `local` as -/// `sb_library_open` takes it. While the provider can't be reached, a folder listed before +/// `sb_library_open` takes it, and `device` naming this device on the conflict pages its own +/// edits make when another device's upload replaced them. While the provider can't be reached, a folder listed before /// opens as last listed, and sections opened before open from their copies; a revoked grant /// refuses to open and keeps them. /// /// # Safety -/// `root` and `cache` are NUL-terminated UTF-8; `error` is null or a place for a string. +/// `root`, `device` and `cache` are NUL-terminated UTF-8; `error` is null or a place for a +/// string. #[unsafe(no_mangle)] pub unsafe extern "C" fn sb_library_provider( root: *const c_char, folder: bool, local: bool, + device: *const c_char, cache: *const c_char, error: *mut *mut c_char, ) -> *const Library { + let cache = string(cache); shared( - Library::provider(&string(root), folder, local, Path::new(&string(cache))), + Library::provider( + &string(root), + folder, + local, + &string(device), + Path::new(&cache), + ), error, ) } diff --git a/crates/mobile/src/provider.rs b/crates/mobile/src/provider.rs index 67139ddb07a625741739b4422adc38457d98377e..395da6af0cb781827cfee96527e3a81e7a57b009 100644 --- a/crates/mobile/src/provider.rs +++ b/crates/mobile/src/provider.rs @@ -35,7 +35,7 @@ pub struct Table { pub document: extern "C" fn(id: *const c_char, found: Found, context: *mut c_void) -> i32, /// Calls `found` for each document in the folder `id`. pub children: extern "C" fn(id: *const c_char, found: Found, context: *mut c_void) -> i32, - /// Opens `id` in `ContentResolver` mode `r` or `rw`, setting `fd` to a descriptor the + /// Opens `id` in a `ContentResolver` mode, setting `fd` to a descriptor the /// caller owns and closes, and `handle` to what `close` finishes. pub open: extern "C" fn( id: *const c_char, diff --git a/crates/mobile/src/tests.rs b/crates/mobile/src/tests.rs index ecdf5934b2ec5d01830a33f73f1c6a999537b671..e827e86d4f9a45254e73c16f3fb33242a1f8e7f1 100644 --- a/crates/mobile/src/tests.rs +++ b/crates/mobile/src/tests.rs @@ -3752,9 +3752,16 @@ fn provided(root: &Path, folder: bool, cache: &Path) -> std::result::Result io::Result; fn children(&self, id: &str) -> io::Result>; - /// Opens a document in a `ContentResolver` mode, `r` or `rw`. + /// Opens a document in a `ContentResolver` mode: `r`, `rw`, `wt` or `w`. fn open(&self, id: &str, mode: &str) -> io::Result; /// Closes a file `open` gave, with the provider's verdict on what was written to it. fn close(&self, file: File) -> io::Result<()>; @@ -46,15 +47,26 @@ pub trait Provider: Send + Sync { pub struct Documents { provider: Box, pub(crate) root: String, + /// The app's cache, holding the replicas and the notebook's listing. + pub(crate) cache: PathBuf, + /// This device as the conflict pages its merged-back publications make name it. + device: String, /// Each path's document as last listed; a stale one is listed again. ids: Mutex>, } impl Documents { - pub fn new(provider: impl Provider + 'static, root: impl Into) -> Self { + pub fn new( + provider: impl Provider + 'static, + root: impl Into, + cache: impl Into, + device: impl Into, + ) -> Self { Self { provider: Box::new(provider), root: root.into(), + cache: cache.into(), + device: device.into(), ids: Mutex::default(), } } @@ -63,6 +75,12 @@ impl Documents { format!("provider:{}", self.root) } + /// Where this device keeps the images it published until the documents are seen holding + /// them: a provider without compare-and-swap may let another device's upload replace one. + fn published(&self) -> PathBuf { + crate::location::folder(&self.cache, &self.location()).join("published") + } + fn id(&self, path: &str) -> io::Result { if path.is_empty() { return Ok(self.root.clone()); @@ -150,9 +168,12 @@ impl Documents { self.opened(id, "r", |file| { if file.stream_position().is_err() { let image = whole(file, crate::MAX_FILE_BYTES)?; - let stamp = Stamp::of(&image) - .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; - return Ok((stamp, Some(image))); + // A torn image shorter than a header still stamps, as it stands. + let mut header = [0; 1024]; + let start = &image[..image.len().min(1024)]; + header[..start.len()].copy_from_slice(start); + let length = image.len() as u64; + return Ok((Stamp { header, length }, Some(image))); } let mut header = [0; 1024]; file.read_exact(&mut header)?; @@ -162,42 +183,80 @@ impl Documents { }) } - /// Writes the document at `path` in place through `operation`, provided its grant allows - /// writing and it opens seekable. The id is the one the last listing gave, never listed - /// again, so that a document taking the name meanwhile is not the one written. + /// Writes the document at `path` through `operation`, provided its grant allows writing: + /// in place where its `r` descriptor seeks, else on the image read whole, which then + /// replaces it. `keep` takes what was written: the descriptor, once the commit landed, or + /// the image, once a write to the document opened. The id is the one the last listing gave, + /// never listed again, so that a document taking the name meanwhile is not the one written. fn write( &self, path: &str, - operation: impl FnOnce(&mut Seekable) -> std::result::Result<(), CommitError>, + operation: impl FnOnce(&mut Io) -> std::result::Result<(), CommitError>, + keep: impl FnOnce(Written<'_>) -> io::Result<()>, ) -> std::result::Result<(), CommitError> { - let refused = |error| CommitError { - state: CommitState::NotCommitted, - error, - }; - let read_only = |reason| refused(io::Error::new(io::ErrorKind::ReadOnlyFilesystem, reason)); let id = self.id(path).map_err(refused)?; - if !self.provider.document(&id).map_err(refused)?.writable { - return Err(read_only( - "The app keeping this section lets Snowbound only read it", - )); - } + self.writable(&id).map_err(refused)?; let seekable = self .opened(&id, "r", |file| Ok(file.stream_position().is_ok())) .map_err(refused)?; - // Streamed documents stay read-only until lost publications merge back (plan commit 2). - if !seekable { - return Err(read_only( - "The app keeping this section can't save changes in place yet; they wait on this device", - )); - } - let mut io = Seekable(self.provider.open(&id, "rw").map_err(refused)?); + let mut io = if seekable { + Io::File(self.provider.open(&id, "rw").map_err(refused)?) + } else { + let image = self.opened(&id, "r", |file| whole(file, crate::MAX_FILE_BYTES)); + Io::Image(image.map_err(refused)?) + }; let result = operation(&mut io); - let closed = self.provider.close(io.0); - result?; - closed.map_err(|error| CommitError { - state: CommitState::Unknown, - error, - }) + match io { + Io::File(mut file) => { + // A copy it could not keep only leaves the publication unprotected. + if result.is_ok() { + let _ = keep(Written::InPlace(&mut file)); + } + let closed = self.provider.close(file); + result?; + closed.map_err(uncertain) + } + Io::Image(image) => { + result?; + self.upload(&id, &image, keep) + } + } + } + + /// Fails unless the grant allows writing the document `id`. + fn writable(&self, id: &str) -> io::Result<()> { + if self.provider.document(id)?.writable { + return Ok(()); + } + Err(io::Error::new( + io::ErrorKind::ReadOnlyFilesystem, + "The app keeping this section lets Snowbound only read it", + )) + } + + /// Writes `image` whole as the document `id`, with `wt`, or `w` where that is refused + /// (images only grow), having `keep` take it once a write opened. + fn upload( + &self, + id: &str, + image: &[u8], + keep: impl FnOnce(Written<'_>) -> io::Result<()>, + ) -> std::result::Result<(), CommitError> { + // A provider refusing `wt` may say so as any failure but a revoked grant. + let mut file = match self.provider.open(id, "wt") { + Err(error) if error.kind() != io::ErrorKind::PermissionDenied => { + self.provider.open(id, "w") + } + opened => opened, + } + .map_err(refused)?; + if let Err(error) = keep(Written::Image(image)) { + let _ = self.provider.close(file); + return Err(refused(error)); + } + let written = file.write_all(image); + let closed = self.provider.close(file); + written.and(closed).map_err(uncertain) } /// Renames the document at `path` within its folder. @@ -208,6 +267,20 @@ impl Documents { } } +fn refused(error: io::Error) -> CommitError { + CommitError { + state: CommitState::NotCommitted, + error, + } +} + +fn uncertain(error: io::Error) -> CommitError { + CommitError { + state: CommitState::Unknown, + error, + } +} + fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { mutex.lock().unwrap_or_else(PoisonError::into_inner) } @@ -221,30 +294,116 @@ fn whole(file: &mut File, limit: usize) -> io::Result> { Ok(bytes) } -/// A seekable descriptor as a commit writes it. -struct Seekable(File); +/// What a write wrote: the document in place, or its whole image. +enum Written<'a> { + InPlace(&'a mut File), + Image(&'a [u8]), +} -impl CommitIo for Seekable { +/// A document as a commit writes it: its seekable descriptor, or its image to write whole. +enum Io { + File(File), + Image(Vec), +} + +impl CommitIo for Io { fn read_at(&mut self, offset: u64, bytes: &mut [u8]) -> io::Result { - self.0.seek(SeekFrom::Start(offset))?; - self.0.read(bytes) + match self { + Io::File(file) => { + file.seek(SeekFrom::Start(offset))?; + file.read(bytes) + } + Io::Image(image) => { + let rest = usize::try_from(offset) + .ok() + .and_then(|offset| image.get(offset..)) + .unwrap_or_default(); + let count = rest.len().min(bytes.len()); + bytes[..count].copy_from_slice(&rest[..count]); + Ok(count) + } + } } fn write_at(&mut self, offset: u64, bytes: &[u8]) -> io::Result { - self.0.seek(SeekFrom::Start(offset))?; - self.0.write(bytes) + match self { + Io::File(file) => { + file.seek(SeekFrom::Start(offset))?; + file.write(bytes) + } + Io::Image(image) => { + let offset = usize::try_from(offset).map_err(|_| io::ErrorKind::FileTooLarge)?; + let end = offset + bytes.len(); + if image.len() < end { + image.resize(end, 0); + } + image[offset..end].copy_from_slice(bytes); + Ok(bytes.len()) + } + } } fn flush(&mut self) -> io::Result<()> { - self.0.sync_all() + match self { + Io::File(file) => file.sync_all(), + Io::Image(_) => Ok(()), + } } } -/// A section a document provider keeps, as its replica publishes to it. +/// The hex of a stamp's digest, which names an image this device keeps. +fn digest(stamp: &Stamp) -> String { + use sha2::Digest; + let digest = sha2::Sha256::new() + .chain_update(stamp.header) + .chain_update(stamp.length.to_le_bytes()) + .finalize(); + hex(&digest[..16]) +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} + +fn file_of(header: &[u8]) -> Option<[u8; 16]> { + Some(onestore::Header::parse(header).ok()?.file_id) +} + +/// An image this device keeps, as its name gives it: `--.one`, the +/// document's identity, then the digests of the image's stamp and of the stamp of the image +/// it was published on. +struct Kept { + name: String, + document: String, + image: String, + base: String, +} + +impl Kept { + fn named(name: String) -> Option { + let mut parts = name.strip_suffix(".one")?.split('-').map(str::to_owned); + let (document, image, base) = (parts.next()?, parts.next()?, parts.next()?); + Some(Self { + name, + document, + image, + base, + }) + } +} + +/// How long reads may lag this device's own upload before a document still standing as the +/// image the upload was made on counts as having lost it. +const LAG: Duration = Duration::from_secs(60); + +/// A section a document provider keeps, as its replica publishes to it. Each image it +/// publishes is kept, and listed as a version once the document is seen otherwise, until the +/// document holds it: an upload another device's replaced then merges back as a conflict +/// version does (`resolve.rs`). Only this device's publications are kept so. pub(crate) struct ProviderRemote { documents: Arc, path: String, - /// The image the last stamp had to read whole, which the step's read takes. + /// The image the step read whole for a stamp, which its read takes. read: Option>, } @@ -256,6 +415,82 @@ impl ProviderRemote { read: None, } } + + fn kept(&self) -> Vec<(Kept, std::time::SystemTime)> { + let entries = crate::fs::read_dir(self.documents.published()) + .into_iter() + .flatten(); + entries + .filter_map(|entry| { + let entry = entry.ok()?; + let kept = Kept::named(entry.file_name().into_string().ok()?)?; + Some((kept, entry.metadata().ok()?.modified().ok()?)) + }) + .collect() + } + + /// Keeps the image `transaction` published, in place of the one it was published on: + /// that kept image with the transaction applied where this device kept it, else read back. + fn keep(&self, written: Written<'_>, transaction: &Transaction) -> io::Result<()> { + let folder = self.documents.published(); + let base = digest(transaction.base()); + let document = file_of(&transaction.base().header).ok_or(io::ErrorKind::InvalidData)?; + let on = (self.kept().into_iter()) + .find(|(kept, _)| kept.document == hex(&document) && kept.image == base); + let image = match (written, &on) { + (Written::Image(image), _) => image.to_vec(), + (Written::InPlace(_), Some((kept, _))) => { + let mut image = crate::fs::read(folder.join(&kept.name))?; + transaction + .apply(&mut image) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + image + } + (Written::InPlace(file), None) => { + file.seek(SeekFrom::Start(0))?; + whole(file, crate::MAX_FILE_BYTES)? + } + }; + let name = format!( + "{}-{}-{base}.one", + hex(&document), + digest(&transaction.stamp()) + ); + crate::fs::create_dir_all(&folder)?; + let mut kept = tempfile::NamedTempFile::new_in(&folder)?; + kept.write_all(&image)?; + kept.as_file().sync_all()?; + kept.persist(folder.join(name)) + .map_err(|error| error.error)?; + if let Some((kept, _)) = on { + crate::fs::remove_file(folder.join(kept.name))?; + } + Ok(()) + } + + /// This device's own upload cut off, as a provider that writes non-atomically leaves it, + /// put back and read again: the newest image kept that the torn bytes begin. Any other + /// document that does not parse is left as it is. + fn repair(&self, torn: &[u8]) -> io::Result> { + let mut newest = None; + for (kept, at) in self.kept() { + let image = crate::fs::read(self.documents.published().join(&kept.name))?; + if image.len() > torn.len() && image.starts_with(torn) { + newest = newest.max(Some((at, image))); + } + } + let (_, image) = newest.ok_or(io::ErrorKind::WouldBlock)?; + let id = self.documents.id(&self.path)?; + self.documents.writable(&id)?; + self.documents + .upload(&id, &image, |_| Ok(())) + .map_err(|error| error.error)?; + self.documents.read(&self.path, crate::MAX_FILE_BYTES) + } +} + +fn parses(image: &[u8]) -> bool { + Store::parse(image).is_ok_and(|store| store.checksum_mismatches.is_empty()) } impl crate::Remote for ProviderRemote { @@ -265,7 +500,7 @@ impl crate::Remote for ProviderRemote { Some(image) => image, None => self.documents.read(&self.path, crate::MAX_FILE_BYTES)?, }; - if !Store::parse(&image).is_ok_and(|store| store.checksum_mismatches.is_empty()) { + if !parses(&image) { return Err(io::Error::new( io::ErrorKind::WouldBlock, "The section was read mid-write", @@ -275,6 +510,10 @@ impl crate::Remote for ProviderRemote { } fn stamp(&mut self) -> io::Result { + if let Some(image) = &self.read { + return Stamp::of(image) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)); + } let (stamp, image) = self.documents.stamp(&self.path)?; self.read = image; Ok(stamp) @@ -282,14 +521,63 @@ impl crate::Remote for ProviderRemote { fn publish(&mut self, transaction: &Transaction) -> std::result::Result<(), CommitError> { self.read = None; - self.documents - .write(&self.path, |io| transaction.commit(io)) + self.documents.write( + &self.path, + |io| transaction.commit(io), + |written| self.keep(written, transaction), + ) } fn confirm(&mut self, base: &Stamp) -> std::result::Result<(), CommitError> { self.read = None; self.documents - .write(&self.path, |io| onestore::confirm(io, base)) + .write(&self.path, |io| onestore::confirm(io, base), |_| Ok(())) + } + + /// The images kept of this document other than the one it stands as; one whose upload + /// the document still stands before is listed only once reads have had `LAG` to catch up. + /// A document read torn is repaired first, once a step. + fn versions(&mut self) -> io::Result> { + let (mut stamp, mut image) = self.documents.stamp(&self.path)?; + if let Some(torn) = image.as_ref().filter(|image| !parses(image)) + && let Ok(repaired) = self.repair(torn) + && let Ok(repaired_stamp) = Stamp::of(&repaired) + { + (stamp, image) = (repaired_stamp, Some(repaired)); + } + self.read = image; + // A header mid-write names no document; the step's read repairs it. + let Some(document) = file_of(&stamp.header) else { + return Ok(Vec::new()); + }; + let (document, current) = (hex(&document), digest(&stamp)); + let lagging = |kept: &Kept, at: std::time::SystemTime| { + kept.base == current && at.elapsed().is_ok_and(|elapsed| elapsed < LAG) + }; + Ok((self.kept().into_iter()) + .filter(|(kept, at)| { + kept.document == document && kept.image != current && !lagging(kept, *at) + }) + .map(|(kept, _)| crate::Version { + id: kept.name, + device: Some(self.documents.device.clone()), + }) + .collect()) + } + + fn version(&mut self, id: &str) -> io::Result> { + crate::fs::read(self.documents.published().join(id)) + } + + /// One that cannot merge, of another section under the document's name, is set aside + /// whole, where `versions` lists it no more: no provider here takes a new document beside it. + fn retire(&mut self, id: &str, keep: bool) -> io::Result<()> { + let path = self.documents.published().join(id); + if keep { + crate::fs::rename(&path, path.with_extension("unmerged")) + } else { + crate::fs::remove_file(path) + } } } @@ -439,20 +727,24 @@ impl Storage for Tree { } fn place(&self, path: &str, ancestor: [u8; 16], name: &str) -> Result<()> { - Ok(self.documents.write(path, |io| { + let place = |io: &mut Io| { onestore::place(io, ancestor, name).map_err(|error| CommitError { state: CommitState::Unknown, error, }) - })?) + }; + Ok(self.documents.write(path, place, |_| Ok(()))?) } fn commit(&self, path: &str, transaction: &Transaction) -> Result<()> { - Ok(self.documents.write(path, |io| transaction.commit(io))?) + Ok(self + .documents + .write(path, |io| transaction.commit(io), |_| Ok(()))?) } fn confirm(&self, path: &str, base: &Stamp) -> std::result::Result<(), CommitError> { - self.documents.write(path, |io| onestore::confirm(io, base)) + self.documents + .write(path, |io| onestore::confirm(io, base), |_| Ok(())) } fn supersede(&self, _: &str, _: &Stamp, _: &str) -> Result<()> { diff --git a/crates/notebook/src/provider/tests.rs b/crates/notebook/src/provider/tests.rs index e5a1320f95362768c1f1bdbf8efe6cff99c11141..00aebee6d2472fa00e79885fe6093654e7e34204 100644 --- a/crates/notebook/src/provider/tests.rs +++ b/crates/notebook/src/provider/tests.rs @@ -12,9 +12,10 @@ use onestore::{ page::{Page, PageObject}, }; use std::{ + collections::HashMap, fs, ops::Range, - os::fd::OwnedFd, + os::fd::{AsRawFd, OwnedFd}, path::Path, thread, time::{Duration, Instant}, @@ -25,8 +26,10 @@ const TOC: &str = "Open Notebook.onetoc2"; #[derive(Clone, Copy, PartialEq)] enum Kind { Seekable, - /// Pipes in every mode, `rw` included, with no metadata. - Stream, + /// Pipes in every mode, `rw` included, with no metadata; `truncate` where `wt` is taken. + Stream { + truncate: bool, + }, } struct Node { @@ -50,9 +53,19 @@ struct State { /// The next `rw` open finds its document renamed to `Moved.one` under a new id, and a /// document holding these bytes in its place. swap: Option>, + /// The next stream upload fails after this many bytes, leaving them as the document where + /// `.1`, as a provider that writes non-atomically does, or else the document as it was. + cut: Option<(u64, bool)>, + /// Streams refuse to open for writing, as a provider out of space does. + full: bool, modes: Vec, + /// Each stream being uploaded, by descriptor: its document, whether `wt` truncates it, and + /// what arrived with the cut that stopped it, if one did. + uploads: HashMap, } +type Upload = thread::JoinHandle<(Vec, Option<(u64, bool)>)>; + struct Shared { folder: tempfile::TempDir, state: Mutex, @@ -86,7 +99,10 @@ impl Fake { revoked: false, delayed: false, swap: None, + cut: None, + full: false, modes: Vec::new(), + uploads: HashMap::new(), }; Self { store: Arc::new(Shared { @@ -105,7 +121,13 @@ impl Fake { } fn documents(&self, root: &str) -> Arc { - Arc::new(Documents::new(self.clone(), root)) + let cache = self + .store + .folder + .path() + .join(format!("cache-{}", self.device)); + let device = format!("Device {}", self.device); + Arc::new(Documents::new(self.clone(), root, cache, device)) } fn state(&self) -> MutexGuard<'_, State> { @@ -300,18 +322,59 @@ impl Provider for Fake { .truncate(mode == "wt") .open(file) } - (Kind::Stream, "r") => { + (Kind::Stream { .. }, "r") => { let (reader, mut writer) = io::pipe()?; let bytes = fs::read(self.file(id))?; thread::spawn(move || writer.write_all(&bytes)); Ok(File::from(OwnedFd::from(reader))) } - (Kind::Stream, _) => Ok(File::from(OwnedFd::from(io::pipe()?.1))), + (Kind::Stream { .. }, "rw") => Ok(File::from(OwnedFd::from(io::pipe()?.1))), + // As Android's host reports a refused mode: unreachable, like most failures. + (Kind::Stream { truncate: false }, "wt") => Err(io::Error::new( + io::ErrorKind::NotConnected, + "Unsupported mode", + )), + (Kind::Stream { .. }, _) if state.full => Err(io::Error::other("No space left")), + (Kind::Stream { .. }, _) => { + let (reader, writer) = io::pipe()?; + let cut = state.cut.take(); + let upload = thread::spawn(move || { + let mut received = Vec::new(); + let at = cut.map_or(u64::MAX, |(at, _)| at); + let _ = reader.take(at).read_to_end(&mut received); + (received, cut) + }); + let writer = File::from(OwnedFd::from(writer)); + let upload = (id.to_owned(), mode == "wt", upload); + state.uploads.insert(writer.as_raw_fd(), upload); + Ok(writer) + } } } - fn close(&self, _: File) -> io::Result<()> { - Ok(()) + fn close(&self, file: File) -> io::Result<()> { + let upload = self.state().uploads.remove(&file.as_raw_fd()); + drop(file); + let Some((id, truncate, upload)) = upload else { + return Ok(()); + }; + let (received, cut) = upload.join().unwrap(); + if let Some((_, torn)) = cut { + if torn { + fs::write(self.file(&id), received)?; + } + return Err(io::Error::other("The upload failed")); + } + let mut bytes = if truncate { + Vec::new() + } else { + fs::read(self.file(&id))? + }; + if bytes.len() < received.len() { + bytes.resize(received.len(), 0); + } + bytes[..received.len()].copy_from_slice(&received); + fs::write(self.file(&id), bytes) } fn rename(&self, id: &str, name: &str) -> io::Result { @@ -326,10 +389,33 @@ impl Provider for Fake { } } +/// A section holding `text`, edited once since it was made, as a section OneNote has written +/// is: the first edit of a page `create_section` makes starts a fresh revision chain, so its +/// pages' later states trace back to that edit, not to the section as made. +fn section_image(name: &str, text: &str) -> Vec { + let (first, rest) = text.split_at(1); + let mut image = onestore::create_section(name, rest, "Author").unwrap(); + let arena = onestore::Arena::default(); + let mut section = onestore::Section::open(&arena, image.clone()).unwrap(); + let space = section.pages().unwrap()[0].0; + let op = PageOp::Text { + text: paragraph(§ion.page(space).unwrap()).0, + range: 0..0, + with: first.into(), + }; + let edit = Edit { + at: crate::now(), + ops: vec![Op::Page { space, op }], + }; + section.apply("Author", &edit).unwrap(); + section.seal().unwrap().unwrap().apply(&mut image).unwrap(); + image +} + /// A notebook folder: its TOC and two sections it lists, placed under it. fn notebook(fake: &Fake) { - let mut notes = onestore::create_section("Notes.one", "Original text", "Author").unwrap(); - let mut ideas = onestore::create_section("Ideas.one", "First idea", "Author").unwrap(); + let mut notes = section_image("Notes.one", "Original text"); + let mut ideas = section_image("Ideas.one", "First idea"); let file = |image: &[u8]| Store::parse(image).unwrap().header.file_id; let toc = onestore::create_table_of_contents( TOC, @@ -432,9 +518,8 @@ fn status(section: &session::Section, accept: impl Fn(&SyncStatus) -> bool) -> S fn seekable_documents_take_commits_in_place_and_rebase_on_another_devices() { let fake = Fake::new(Kind::Seekable); notebook(&fake); - let cache = tempfile::tempdir().unwrap(); let documents = fake.documents("root"); - let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]); let section = open(¬ebook, &documents, "Notes.one"); let before = fake.bytes("Notes.one").len(); @@ -446,8 +531,7 @@ fn seekable_documents_take_commits_in_place_and_rebase_on_another_devices() { assert!(fake.modes().iter().all(|mode| mode == "r" || mode == "rw")); let elsewhere = fake.device(1).documents("root"); - let other_cache = tempfile::tempdir().unwrap(); - let other = Notebook::open_provider(Arc::clone(&elsewhere), other_cache.path()).unwrap(); + let other = Notebook::open_provider(Arc::clone(&elsewhere)).unwrap(); let other = open(&other, &elsewhere, "Notes.one"); let id = typed(&other, 0..0, "World "); published(&other, id); @@ -467,44 +551,24 @@ fn seekable_documents_take_commits_in_place_and_rebase_on_another_devices() { } #[test] -fn stream_only_documents_are_read_and_never_opened_to_write() { - let fake = Fake::new(Kind::Stream); - notebook(&fake); - let cache = tempfile::tempdir().unwrap(); - let documents = fake.documents("root"); - let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); - let section = open(¬ebook, &documents, "Notes.one"); - let id = typed(§ion, 13..13, " mine"); - let waiting = status(§ion, |status| status.error.is_some()); - assert_eq!(waiting.state(), SyncState::ReadOnly); - assert_eq!(waiting.queued, 1); - assert!(fake.modes().iter().all(|mode| mode == "r")); - - // Another device, which can write in place, changes the document. - let elsewhere = Fake::new(Kind::Seekable); - for name in [TOC, "Notes.one", "Ideas.one"] { - elsewhere.add(name, &fake.bytes(name)); +fn stream_only_documents_take_whole_images_and_never_open_to_write_in_place() { + for truncate in [true, false] { + let fake = Fake::new(Kind::Stream { truncate }); + notebook(&fake); + let documents = fake.documents("root"); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); + let section = open(¬ebook, &documents, "Ideas.one"); + let id = typed(§ion, 0..0, "Second. "); + published(§ion, id); + let id = typed(§ion, 0..0, "Third. "); + published(§ion, id); + assert_eq!(stored(fake.bytes("Ideas.one")), "Third. Second. First idea"); + let modes = fake.modes(); + let count = |wanted: &str| modes.iter().filter(|mode| *mode == wanted).count(); + assert_eq!(count("rw"), 0, "{modes:?}"); + assert_eq!((count("wt"), count("w")), (2, if truncate { 0 } else { 2 })); + section.close().unwrap(); } - let other_documents = elsewhere.documents("root"); - let other_cache = tempfile::tempdir().unwrap(); - let other = Notebook::open_provider(Arc::clone(&other_documents), other_cache.path()).unwrap(); - let other = open(&other, &other_documents, "Notes.one"); - let theirs = typed(&other, 0..0, "Theirs "); - published(&other, theirs); - other.close().unwrap(); - fs::write( - fake.file(&fake.id("Notes.one")), - elsewhere.bytes("Notes.one"), - ) - .unwrap(); - until("the change was read", || { - section.wake(); - text(§ion) == "Theirs Original text mine" - }); - assert_eq!(section.pending().unwrap()[0].id, id); - assert_eq!(stored(fake.bytes("Notes.one")), "Theirs Original text"); - assert!(fake.modes().iter().all(|mode| mode == "r")); - section.close().unwrap(); } #[test] @@ -512,9 +576,8 @@ fn documents_without_a_write_grant_are_only_read() { let fake = Fake::new(Kind::Seekable); notebook(&fake); fake.set(|state| state.writable = false); - let cache = tempfile::tempdir().unwrap(); let documents = fake.documents("root"); - let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); let section = open(¬ebook, &documents, "Notes.one"); let before = fake.bytes("Notes.one"); typed(§ion, 0..0, "Mine "); @@ -536,9 +599,8 @@ fn a_write_another_device_sees_late_rebases_it_once_visible() { notebook(&fake); fake.set(|state| state.delayed = true); let (writer, reader) = (fake.documents("root"), fake.device(1).documents("root")); - let caches = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap()); - let writing = Notebook::open_provider(Arc::clone(&writer), caches.0.path()).unwrap(); - let reading = Notebook::open_provider(Arc::clone(&reader), caches.1.path()).unwrap(); + let writing = Notebook::open_provider(Arc::clone(&writer)).unwrap(); + let reading = Notebook::open_provider(Arc::clone(&reader)).unwrap(); let (writing, reading) = ( open(&writing, &writer, "Notes.one"), open(&reading, &reader, "Notes.one"), @@ -567,9 +629,8 @@ fn a_write_another_device_sees_late_rebases_it_once_visible() { fn a_cold_start_while_offline_opens_the_last_listing_and_queued_edits() { let fake = Fake::new(Kind::Seekable); notebook(&fake); - let cache = tempfile::tempdir().unwrap(); let documents = fake.documents("root"); - let mut notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let mut notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); let background = Background::provider(Arc::clone(&documents), || {}).unwrap(); background.watch(notebook.replicas()); until("every section has its offline copy", || { @@ -589,7 +650,7 @@ fn a_cold_start_while_offline_opens_the_last_listing_and_queued_edits() { // A new process: the provider fails throughout. let documents = fake.documents("root"); - let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]); let notes = open(¬ebook, &documents, "Notes.one"); assert_eq!(text(¬es), "Original text"); @@ -608,9 +669,8 @@ fn a_cold_start_while_offline_opens_the_last_listing_and_queued_edits() { fn a_revoked_grant_is_not_offline_and_keeps_the_replicas() { let fake = Fake::new(Kind::Seekable); notebook(&fake); - let cache = tempfile::tempdir().unwrap(); let documents = fake.documents("root"); - let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); let replica = notebook.replica_path("Notes.one").unwrap(); let section = open(¬ebook, &documents, "Notes.one"); fake.set(|state| state.revoked = true); @@ -622,7 +682,7 @@ fn a_revoked_grant_is_not_offline_and_keeps_the_replicas() { ); section.close().unwrap(); drop(notebook); - let refused = Notebook::open_provider(fake.documents("root"), cache.path()).err(); + let refused = Notebook::open_provider(fake.documents("root")).err(); assert!( matches!( &refused, @@ -638,7 +698,7 @@ fn a_revoked_grant_is_not_offline_and_keeps_the_replicas() { fake.set(|state| state.revoked = false); let documents = fake.documents("root"); - let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); let section = open(¬ebook, &documents, "Notes.one"); published(§ion, id); 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() { fn documents_renamed_to_new_ids_are_listed_again() { let fake = Fake::new(Kind::Seekable); notebook(&fake); - let cache = tempfile::tempdir().unwrap(); let documents = fake.documents("root"); - let mut notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let mut notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); let section = open(¬ebook, &documents, "Notes.one"); let id = typed(§ion, 0..0, "One "); published(§ion, id); @@ -691,9 +750,8 @@ fn of_documents_sharing_a_name_one_lists_and_the_rest_are_unavailable() { "Notes.one", &onestore::create_section("Notes.one", "A twin", "Author").unwrap(), ); - let cache = tempfile::tempdir().unwrap(); let documents = fake.documents("root"); - let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]); let unavailable = ¬ebook.catalog().unavailable; assert_eq!(unavailable.len(), 1); @@ -707,9 +765,8 @@ fn of_documents_sharing_a_name_one_lists_and_the_rest_are_unavailable() { fn a_document_taking_the_name_between_check_and_write_is_not_written() { let fake = Fake::new(Kind::Seekable); notebook(&fake); - let cache = tempfile::tempdir().unwrap(); let documents = fake.documents("root"); - let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); let section = open(¬ebook, &documents, "Notes.one"); let id = typed(§ion, 0..0, "Checked "); published(§ion, id); @@ -761,9 +818,8 @@ fn conflicting_edits_leave_a_conflict_page_in_the_document() { let fake = Fake::new(Kind::Seekable); notebook(&fake); let (mine, theirs) = (fake.documents("root"), fake.device(1).documents("root")); - let caches = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap()); - let local = Notebook::open_provider(Arc::clone(&mine), caches.0.path()).unwrap(); - let remote = Notebook::open_provider(Arc::clone(&theirs), caches.1.path()).unwrap(); + let local = Notebook::open_provider(Arc::clone(&mine)).unwrap(); + let remote = Notebook::open_provider(Arc::clone(&theirs)).unwrap(); let (local, remote) = ( open(&local, &mine, "Notes.one"), open(&remote, &theirs, "Notes.one"), @@ -793,10 +849,8 @@ fn a_lone_document_opens_offline_from_its_replica() { let fake = Fake::new(Kind::Seekable); let image = onestore::create_section("Lone.one", "Alone", "Author").unwrap(); let lone = fake.add("Lone.one", &image); - let cache = tempfile::tempdir().unwrap(); - let open = |fake: &Fake| { - session::Section::open_provider(fake.documents(&lone), cache.path(), false, || {}).unwrap() - }; + let open = + |fake: &Fake| session::Section::open_provider(fake.documents(&lone), false, || {}).unwrap(); let section = open(&fake); let id = typed(§ion, 0..0, "Not "); published(§ion, id); @@ -817,15 +871,13 @@ fn a_lone_document_opens_offline_from_its_replica() { fn a_provider_that_lists_but_cannot_open_opens_sections_from_their_replicas() { let fake = Fake::new(Kind::Seekable); notebook(&fake); - let cache = tempfile::tempdir().unwrap(); let documents = fake.documents("root"); - let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); open(¬ebook, &documents, "Notes.one").close().unwrap(); drop(notebook); // Another device changes the section, so it lists otherwise than when last read. let elsewhere = fake.device(1).documents("root"); - let other_cache = tempfile::tempdir().unwrap(); - let other = Notebook::open_provider(Arc::clone(&elsewhere), other_cache.path()).unwrap(); + let other = Notebook::open_provider(Arc::clone(&elsewhere)).unwrap(); let other = open(&other, &elsewhere, "Notes.one"); let id = typed(&other, 0..0, "Theirs "); published(&other, id); @@ -833,7 +885,7 @@ fn a_provider_that_lists_but_cannot_open_opens_sections_from_their_replicas() { fake.set(|state| state.unopenable = true); let documents = fake.documents("root"); - let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]); let section = open(¬ebook, &documents, "Notes.one"); assert_eq!(text(§ion), "Original text"); @@ -846,3 +898,273 @@ fn a_provider_that_lists_but_cannot_open_opens_sections_from_their_replicas() { assert_eq!(stored(fake.bytes("Notes.one")), "Theirs Original text mine"); section.close().unwrap(); } + +/// The images a device keeps of what it published, for every notebook in its cache. +fn kept_images(fake: &Fake) -> Vec { + let cache = fake + .store + .folder + .path() + .join(format!("cache-{}", fake.device)); + let folders = fs::read_dir(cache.join("replicas")).into_iter().flatten(); + folders + .filter_map(|folder| fs::read_dir(folder.ok()?.path().join("published")).ok()) + .flatten() + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .filter(|name| name.ends_with(".one")) + .collect() +} + +/// The images a device keeps that Notes.one does not stand as, still to merge. +fn unmerged(fake: &Fake) -> Vec { + let current = digest(&Stamp::of(&fake.bytes("Notes.one")).unwrap()); + let mut images = kept_images(fake); + images.retain(|name| Kept::named(name.clone()).is_some_and(|kept| kept.image != current)); + images +} + +#[test] +fn a_publication_another_devices_upload_replaced_merges_back() { + let fake = Fake::new(Kind::Seekable); + notebook(&fake); + fake.set(|state| state.delayed = true); + let (mine, theirs) = (fake.documents("root"), fake.device(1).documents("root")); + let (local, remote) = ( + Notebook::open_provider(Arc::clone(&mine)).unwrap(), + Notebook::open_provider(Arc::clone(&theirs)).unwrap(), + ); + let (local, remote) = ( + open(&local, &mine, "Notes.one"), + open(&remote, &theirs, "Notes.one"), + ); + let id = typed(&local, 0..0, "Alpha "); + published(&local, id); + let id = typed(&remote, 13..13, " beta"); + published(&remote, id); + // Both uploads land; the later one, which never saw the first, is the document. + fake.publish_staged(); + assert_eq!(stored(fake.bytes("Notes.one")), "Original text beta"); + until("the lost publication merged back", || { + local.wake(); + fake.publish_staged(); + stored(fake.bytes("Notes.one")) == "Alpha Original text beta" + }); + until("both devices show it", || { + local.wake(); + remote.wake(); + text(&local) == "Alpha Original text beta" && text(&remote) == text(&local) + }); + assert!(local.conflicts().unwrap().is_empty()); + assert!(unmerged(&fake).is_empty()); + local.close().unwrap(); + remote.close().unwrap(); +} + +/// Ages the images a device keeps past `LAG`, as if reads had long had time to catch up. +fn aged(fake: &Fake) { + let cache = fake + .store + .folder + .path() + .join(format!("cache-{}", fake.device)); + for folder in fs::read_dir(cache.join("replicas")).unwrap() { + for image in fs::read_dir(folder.unwrap().path().join("published")) + .into_iter() + .flatten() + { + let file = File::options() + .write(true) + .open(image.unwrap().path()) + .unwrap(); + let past = std::time::SystemTime::now() - LAG - Duration::from_secs(1); + file.set_modified(past).unwrap(); + } + } +} + +fn failed_attempt(section: &session::Section) { + until("the upload failed", || { + section.wake(); + (section.events().iter()).any(|event| matches!(event, Event::Unreachable(_))) + }); +} + +#[test] +fn an_upload_the_provider_dropped_merges_back_once_reads_could_have_caught_up() { + let fake = Fake::new(Kind::Stream { truncate: true }); + notebook(&fake); + let documents = fake.documents("root"); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); + let section = open(¬ebook, &documents, "Notes.one"); + fake.set(|state| state.cut = Some((2000, false))); + let id = typed(§ion, 0..0, "Cut "); + failed_attempt(§ion); + // The document stands as the upload was made on: reads may lag it, so it waits. + for _ in 0..5 { + section.wake(); + thread::sleep(Duration::from_millis(50)); + } + assert_eq!(kept_images(&fake).len(), 1); + assert_eq!(stored(fake.bytes("Notes.one")), "Original text"); + aged(&fake); + published(§ion, id); + assert_eq!(stored(fake.bytes("Notes.one")), "Cut Original text"); + section.close().unwrap(); +} + +#[test] +fn a_torn_upload_on_a_provider_writing_in_place_is_put_back() { + for at in [2000, 500] { + let fake = Fake::new(Kind::Stream { truncate: true }); + notebook(&fake); + let documents = fake.documents("root"); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); + let section = open(¬ebook, &documents, "Notes.one"); + fake.set(|state| state.cut = Some((at, true))); + let id = typed(§ion, 0..0, "Torn "); + published(§ion, id); + assert_eq!(stored(fake.bytes("Notes.one")), "Torn Original text"); + until("the kept image was retired", || { + section.wake(); + unmerged(&fake).is_empty() + }); + section.close().unwrap(); + } +} + +#[test] +fn an_upload_refused_before_it_began_is_retried_not_merged() { + let fake = Fake::new(Kind::Stream { truncate: true }); + notebook(&fake); + let documents = fake.documents("root"); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); + let section = open(¬ebook, &documents, "Notes.one"); + fake.set(|state| state.full = true); + let id = typed(§ion, 0..0, "Once "); + failed_attempt(§ion); + assert!(kept_images(&fake).is_empty()); + assert_eq!(section.pending().unwrap().len(), 1); + fake.set(|state| state.full = false); + published(§ion, id); + assert_eq!(stored(fake.bytes("Notes.one")), "Once Original text"); + section.close().unwrap(); +} + +#[test] +fn devices_whose_uploads_replace_each_other_converge_with_every_string_once() { + for seed in 1..=4u64 { + let fake = Fake::new(Kind::Seekable); + notebook(&fake); + fake.set(|state| state.delayed = true); + let devices: Vec<_> = (0..3) + .map(|device| { + let documents = fake.device(device).documents("root"); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); + open(¬ebook, &documents, "Notes.one") + }) + .collect(); + let mut random = seed; + let mut next = |bound: u64| { + random = random + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + (random >> 33) % bound + }; + let mut strings = Vec::new(); + for round in 0..6 { + let device = next(3) as usize; + let string = format!("s{seed}r{round}d{device}"); + typed(&devices[device], 0..0, &format!("{string} ")); + strings.push(string); + devices[device].wake(); + thread::sleep(Duration::from_millis(next(150))); + if next(2) == 0 { + fake.publish_staged(); + } + } + until("every device converged", || { + for device in &devices { + device.wake(); + } + thread::sleep(Duration::from_millis(100)); + fake.publish_staged(); + let file = stored(fake.bytes("Notes.one")); + devices + .iter() + .all(|device| device.pending().unwrap().is_empty() && text(device) == file) + && (0..3).all(|device| unmerged(&fake.device(device)).is_empty()) + }); + let arena = onestore::Arena::default(); + let mut section = onestore::Section::open(&arena, fake.bytes("Notes.one")).unwrap(); + let mut conflicts = String::new(); + for (_, pages) in section.conflicts().unwrap() { + for conflict in pages { + conflicts += ¶graph(§ion.page(conflict.space).unwrap()).1; + } + } + let file = stored(fake.bytes("Notes.one")); + for string in &strings { + let shown = file.matches(string.as_str()).count(); + assert!( + shown <= 1, + "seed {seed}: {string} shows {shown} times in {file}" + ); + assert!( + shown == 1 || conflicts.contains(string.as_str()), + "seed {seed}: {string} is gone from {file}" + ); + } + for device in devices { + device.close().unwrap(); + } + } +} + +#[test] +fn a_document_this_device_did_not_write_that_does_not_parse_is_never_rewritten() { + let fake = Fake::new(Kind::Stream { truncate: true }); + notebook(&fake); + let documents = fake.documents("root"); + let notebook = Notebook::open_provider(Arc::clone(&documents)).unwrap(); + let section = open(¬ebook, &documents, "Notes.one"); + let id = typed(§ion, 0..0, "Mine "); + published(§ion, id); + // The same document as another writer took it on, cut off, then with checksums that fail. + let mut theirs = fake.bytes("Notes.one"); + let arena = onestore::Arena::default(); + let mut section_there = onestore::Section::open(&arena, theirs.clone()).unwrap(); + let space = section_there.pages().unwrap()[0].0; + let op = PageOp::Text { + text: paragraph(§ion_there.page(space).unwrap()).0, + range: 0..0, + with: "Theirs ".into(), + }; + let edit = Edit { + at: crate::now(), + ops: vec![Op::Page { space, op }], + }; + section_there.apply("Them", &edit).unwrap(); + let sealed = section_there.seal().unwrap().unwrap(); + sealed.apply(&mut theirs).unwrap(); + let mut damaged = theirs.clone(); + let at = (1024..damaged.len()) + .find(|at| { + let mut flipped = damaged.clone(); + flipped[*at] ^= 0xff; + Store::parse(&flipped).is_ok_and(|store| !store.checksum_mismatches.is_empty()) + }) + .unwrap(); + damaged[at] ^= 0xff; + for unreadable in [theirs[..theirs.len() - 100].to_vec(), damaged] { + fs::write(fake.file(&fake.id("Notes.one")), &unreadable).unwrap(); + let since = fake.modes().len(); + for _ in 0..5 { + section.wake(); + thread::sleep(Duration::from_millis(50)); + } + status(§ion, |status| status.state() == SyncState::InUse); + assert_eq!(fake.bytes("Notes.one"), unreadable); + assert!(fake.modes()[since..].iter().all(|mode| mode == "r")); + } + section.close().unwrap(); +} diff --git a/crates/notebook/src/session.rs b/crates/notebook/src/session.rs index addc69691e51303f449ad567f0c69f79ae6122ad..9b4880989ff1fa5044cc2adbca50c84819b0f769 100644 --- a/crates/notebook/src/session.rs +++ b/crates/notebook/src/session.rs @@ -517,11 +517,9 @@ impl Notebook { /// Opens the notebook whose folder is the root of `documents`. Sections open through /// `Section::resume_provider` with the catalog's paths; while the provider can't be /// reached, the notebook opens as its folders were last listed. - pub fn open_provider( - documents: Arc, - cache: impl AsRef, - ) -> Result { - let listed = listing(cache.as_ref(), &documents.location()).with_extension("entries.json"); + pub fn open_provider(documents: Arc) -> Result { + let cache = documents.cache.clone(); + let listed = listing(&cache, &documents.location()).with_extension("entries.json"); Self::with( Box::new(crate::provider::Tree { documents, listed }), None, @@ -2418,11 +2416,11 @@ impl Section { /// its location has. pub fn open_provider( documents: Arc, - cache: impl AsRef, offline: bool, notify: impl Fn() + Send + Sync + 'static, ) -> Result { - let (cache, location) = (cache.as_ref(), documents.location()); + let (cache, location) = (documents.cache.clone(), documents.location()); + let cache = cache.as_path(); let file = PathBuf::from(&documents.root); let remote = move || Ok(crate::provider::ProviderRemote::new(&documents, "")); let notify = Arc::new(notify); diff --git a/crates/notebook/src/sync.rs b/crates/notebook/src/sync.rs index 111154225d140534d106a027c834d240225a1670..9da7bd631b1b9fe32f73ba4ee8599893ff67e0f8 100644 --- a/crates/notebook/src/sync.rs +++ b/crates/notebook/src/sync.rs @@ -259,7 +259,7 @@ impl Replica { awaited(remote, |remote| remote.publish(&transaction)).await?; false } - // A version this cannot merge is kept whole rather than lost. + // A version this cannot merge is retired to be kept whole, never lost. Ok(Merged::Foreign) | Err(Error::Document(_) | Error::Rejected(_)) => true, Err(error) => return Err(error), };