From c345a48e944a431673a0e8b9d4a59e93dc507a58 Mon Sep 17 00:00:00 2001 From: clover caruso Date: Thu, 15 Oct 2026 13:00:06 -0700 Subject: [PATCH] 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 --- apps/android/jni/snowbound.c | 9 +- .../src/net/paperclover/snowbound/Editor.java | 13 +- crates/mobile/include/snowbound.h | 5 +- crates/mobile/src/library.rs | 23 +- crates/mobile/src/provider.rs | 2 +- crates/mobile/src/tests.rs | 11 +- crates/notebook/README.md | 27 +- crates/notebook/src/provider.rs | 392 ++++++++++++-- crates/notebook/src/provider/tests.rs | 484 +++++++++++++++--- crates/notebook/src/session.rs | 12 +- crates/notebook/src/sync.rs | 2 +- 11 files changed, 816 insertions(+), 164 deletions(-) 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), }; -- 2.54.0