From ffaff48d67462562d069edde11886b41bb411e8c Mon Sep 17 00:00:00 2001 From: clover caruso Date: Thu, 15 Oct 2026 09:16:46 -0700 Subject: [PATCH] feat(notebook): open notebooks through document providers A notebook folder or a lone section that another app keeps, as Android's Files app hands them out, now opens through a document provider the host implements. Sections keep their replicas and queues in the cache as on a share and publish in place where the provider gives a seekable descriptor. A section the provider only streams, or one without a write grant, is read and kept in sync but not written: its edits wait on this device. While the provider cannot be reached, the notebook opens as its folders were last listed and its sections from their offline copies, edits waiting until it is back; a revoked grant keeps every copy. Documents renamed under a new id are found again, a document that takes a section's name between the check and the write is never written, and of documents sharing one name the first lists and the others show as unavailable. Sections can be renamed within their folder; creating, deleting and moving sections and renaming the notebook's folder are not available yet. Refs #24 Assisted-by: claude-opus-5.5 --- crates/notebook/README.md | 27 + crates/notebook/src/background.rs | 19 + crates/notebook/src/discover.rs | 6 +- crates/notebook/src/lib.rs | 1 + crates/notebook/src/provider.rs | 474 +++++++++++++++ crates/notebook/src/provider/tests.rs | 808 ++++++++++++++++++++++++++ crates/notebook/src/session.rs | 139 ++++- crates/notebook/src/session/tests.rs | 1 + 8 files changed, 1459 insertions(+), 16 deletions(-) create mode 100644 crates/notebook/src/provider.rs create mode 100644 crates/notebook/src/provider/tests.rs diff --git a/crates/notebook/README.md b/crates/notebook/README.md index f352974701fc16d330cb0b752b81f18d8c178a6e..b297517f1f9eda97cb4a3729f3be3b0956f9f027 100644 --- a/crates/notebook/README.md +++ b/crates/notebook/README.md @@ -545,6 +545,33 @@ Device and simulator builds link for iOS. Native acceptance uses disposable OneNote 2010 clients and Samba; it does not establish on-device execution or physical power-loss durability. +## Document providers + +`provider` opens notebooks a document provider keeps, as Android's Storage Access Framework +reaches them. The host implements `provider::Provider` (a document's description, a folder's +children, `open` in `ContentResolver`'s `r` or `rw` as an owned `File`, `close` with the provider's +verdict, `rename` returning the new id) and reports an unreachable provider as `NotConnected`, +a revoked grant as `PermissionDenied` and a gone document as `NotFound`. +`provider::Documents::new(provider, root)` holds one root: a picked folder opens +with `Notebook::open_provider(documents, cache)`, its sections with `Section::resume_provider` +and the rest with `Background::provider`, as a Live Share guest's do; a picked document opens +as a lone section with `Section::open_provider`. Catalog paths resolve to ids by listing their +folders, again once when an id has gone, as a rename leaves it; of documents sharing a name the +first by id lists and the rest are `unavailable`, named with their ids. + +A publication resolves the document's id once, from the last listing, and writes nothing to +a document that took its name since. Without a write grant a document is only read, and +publication is refused as `ReadOnlyFilesystem`; where its `r` descriptor seeks, `rw` opens it +and a commit checks the stamp and appends in place exactly as on a disk. A document that only +streams is read whole for each stamp and is read-only too, its edits waiting, until lost +publications merge back. No provider offers a lock or compare-and-swap: the check sees only +this device's view of the document. An image that does not parse is never a base. An open +section checks its file every `Background::UNWATCHED`, the host's change reports waking it +between. While the provider can't be reached the notebook opens as its folders were last +listed and sections from their replicas; a revoked grant keeps them, and the notebook does not +open until access returns. Creating, deleting and moving documents and renaming the folder are +`Unsupported`; renaming within a folder is not. + ## Live presence and Live Share (feature `live`) `live::Live::start(hello, room, reach, relay, events)` listens on a TCP port and, with a diff --git a/crates/notebook/src/background.rs b/crates/notebook/src/background.rs index ebeece4de03bf00a51446767d056f4e532ea19e9..e8fb6834151d73fd98f882b23f178063422f2a32 100644 --- a/crates/notebook/src/background.rs +++ b/crates/notebook/src/background.rs @@ -392,6 +392,25 @@ impl Background { Ok(background) } + /// Keeps the sections of a notebook a document provider keeps in sync while they are not + /// open, with an offline copy of each. Nothing reports the folder's changes; the host + /// passes what it hears of them to `touched`. Paths are catalog paths. + pub fn provider( + documents: Arc, + notify: impl Fn() + Send + 'static, + ) -> Result { + Self::start( + true, + move |_| { + let (bound, listing) = (Arc::clone(&documents), Arc::clone(&documents)); + let bind = move |path: &str| crate::provider::ProviderRemote::new(&bound, path); + let list = move |folder: String| std::future::ready(listing.entries(&folder)); + Ok(((bind, list), false)) + }, + notify, + ) + } + /// Checks what a report names after `settle` rather than a second, where reports come /// one to a commit, as a Live Share host sends them. pub fn set_settle(&self, settle: Duration) { diff --git a/crates/notebook/src/discover.rs b/crates/notebook/src/discover.rs index ceb22da656f51ee3113258e728d1a49e887dbcf9..a91d3e77255ce241605660966b06a7bd2dbf50aa 100644 --- a/crates/notebook/src/discover.rs +++ b/crates/notebook/src/discover.rs @@ -125,7 +125,7 @@ pub struct TocReference { pub order: u32, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] pub enum EntryKind { File, Directory, @@ -135,7 +135,7 @@ pub enum EntryKind { Evicted, } -#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] pub struct Entry { pub name: String, pub kind: EntryKind, @@ -841,7 +841,7 @@ fn component(name: &str) -> bool { !name.is_empty() && name != "." && name != ".." && !name.contains(['/', '\\', '\0']) } -fn join(parent: &str, name: &str) -> String { +pub(crate) fn join(parent: &str, name: &str) -> String { if parent.is_empty() { name.into() } else { diff --git a/crates/notebook/src/lib.rs b/crates/notebook/src/lib.rs index 73398e66c899664b2781125bd79bd7e535da44d2..450ca9d67399a93af8a62bf9f2bd397b23d22232 100644 --- a/crates/notebook/src/lib.rs +++ b/crates/notebook/src/lib.rs @@ -32,6 +32,7 @@ pub mod location; mod merge; mod migrate; pub mod package; +pub mod provider; mod queue; mod recovery; mod resolve; diff --git a/crates/notebook/src/provider.rs b/crates/notebook/src/provider.rs new file mode 100644 index 0000000000000000000000000000000000000000..d3a61566c08a9dae8f178c35855d582e9fbd5f2f --- /dev/null +++ b/crates/notebook/src/provider.rs @@ -0,0 +1,474 @@ +//! Notebooks a document provider keeps, as Android's Storage Access Framework reaches them: +//! documents named by opaque ids, listed by folder and opened as descriptors, with no locks, +//! no rename over another document and, on many providers, no seeking. A picked folder is a +//! notebook root (`session::Notebook::open_provider`), a picked document a lone section +//! (`session::Section::open_provider`); replicas and the catalog's listing stay in the cache. + +use crate::{Error, Result, discover, session::Storage}; +use onestore::{CommitError, CommitIo, CommitState, Stamp, Store, Transaction}; +use std::{ + collections::BTreeMap, + fs::File, + io::{self, Read, Seek, SeekFrom, Write}, + path::PathBuf, + sync::{Arc, Mutex, MutexGuard, PoisonError}, +}; + +/// A document as its provider describes it. +#[derive(Clone, Debug)] +pub struct Document { + pub id: String, + pub name: String, + pub directory: bool, + /// What the provider lists, where it lists anything. + pub size: Option, + pub modified: Option, + /// The grant allows writing and the provider supports it. + pub writable: bool, +} + +/// The host's document provider. Failures take the shapes a provider gives them: unreachable +/// as `NotConnected`, a revoked grant as `PermissionDenied`, and a document gone (or renamed, +/// which changes its id) as `NotFound`. +pub trait Provider: Send + Sync { + fn document(&self, id: &str) -> io::Result; + fn children(&self, id: &str) -> io::Result>; + /// Opens a document in a `ContentResolver` mode, `r` or `rw`. + 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<()>; + /// Renames a document, returning its new id. + fn rename(&self, id: &str, name: &str) -> io::Result; +} + +/// The documents under one root a provider keeps: a notebook's folder or a lone section. +/// Paths are catalog paths, `/`-separated below the root. +pub struct Documents { + provider: Box, + pub(crate) root: 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 { + Self { + provider: Box::new(provider), + root: root.into(), + ids: Mutex::default(), + } + } + + pub(crate) fn location(&self) -> String { + format!("provider:{}", self.root) + } + + fn id(&self, path: &str) -> io::Result { + if path.is_empty() { + return Ok(self.root.clone()); + } + if let Some(id) = lock(&self.ids).get(path) { + return Ok(id.clone()); + } + self.listing(path.rsplit_once('/').map_or("", |(folder, _)| folder))?; + lock(&self.ids) + .get(path) + .cloned() + .ok_or_else(|| io::ErrorKind::NotFound.into()) + } + + /// Reads through the document at `path`, listing its folders again once if the id they + /// last gave is gone. + fn with_id(&self, path: &str, read: impl Fn(&str) -> io::Result) -> io::Result { + match read(&self.id(path)?) { + Err(error) if error.kind() == io::ErrorKind::NotFound && !path.is_empty() => { + lock(&self.ids).clear(); + read(&self.id(path)?) + } + result => result, + } + } + + /// A folder's entries, and the documents sharing a name with one listed before them by id. + fn listing(&self, folder: &str) -> io::Result<(Vec, Vec)> { + let mut children = self.with_id(folder, |id| self.provider.children(id))?; + children.sort_by(|a, b| (&a.name, &a.id).cmp(&(&b.name, &b.id))); + let (mut entries, mut duplicates) = (Vec::::new(), Vec::new()); + let mut ids = lock(&self.ids); + for child in children { + if entries.last().is_some_and(|last| last.name == child.name) { + duplicates.push(child); + continue; + } + ids.insert(discover::join(folder, &child.name), child.id.clone()); + entries.push(discover::Entry { + name: child.name, + kind: if child.directory { + discover::EntryKind::Directory + } else { + discover::EntryKind::File + }, + listed: discover::Listed { + size: child.size.unwrap_or_default(), + // Unknown, it never matches an earlier listing, so the file is read again. + modified: child + .modified + .filter(|_| child.size.is_some()) + .unwrap_or_else(crate::now), + }, + }); + } + Ok((entries, duplicates)) + } + + pub(crate) fn entries(&self, folder: &str) -> io::Result> { + Ok(self.listing(folder)?.0) + } + + /// Opens the document `id`, hands it to `operation` and closes it, failing if either fails. + fn opened( + &self, + id: &str, + mode: &str, + operation: impl FnOnce(&mut File) -> io::Result, + ) -> io::Result { + let mut file = self.provider.open(id, mode)?; + let result = operation(&mut file); + let closed = self.provider.close(file); + let value = result?; + closed?; + Ok(value) + } + + fn read(&self, path: &str, limit: usize) -> io::Result> { + self.with_id(path, |id| self.opened(id, "r", |file| whole(file, limit))) + } + + /// The document's stamp, with its image where a stream had to be read whole for it. + fn stamp(&self, path: &str) -> io::Result<(Stamp, Option>)> { + self.with_id(path, |id| { + 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))); + } + let mut header = [0; 1024]; + file.read_exact(&mut header)?; + let length = file.seek(SeekFrom::End(0))?; + Ok((Stamp { header, length }, None)) + }) + }) + } + + /// 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. + fn write( + &self, + path: &str, + operation: impl FnOnce(&mut Seekable) -> std::result::Result<(), CommitError>, + ) -> 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", + )); + } + 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 result = operation(&mut io); + let closed = self.provider.close(io.0); + result?; + closed.map_err(|error| CommitError { + state: CommitState::Unknown, + error, + }) + } + + /// Renames the document at `path` within its folder. + fn rename(&self, path: &str, name: &str) -> io::Result<()> { + self.provider.rename(&self.id(path)?, name)?; + lock(&self.ids).clear(); + Ok(()) + } +} + +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(PoisonError::into_inner) +} + +fn whole(file: &mut File, limit: usize) -> io::Result> { + let mut bytes = Vec::new(); + file.take(limit as u64 + 1).read_to_end(&mut bytes)?; + if bytes.len() > limit { + return Err(io::ErrorKind::FileTooLarge.into()); + } + Ok(bytes) +} + +/// A seekable descriptor as a commit writes it. +struct Seekable(File); + +impl CommitIo for Seekable { + fn read_at(&mut self, offset: u64, bytes: &mut [u8]) -> io::Result { + self.0.seek(SeekFrom::Start(offset))?; + self.0.read(bytes) + } + + fn write_at(&mut self, offset: u64, bytes: &[u8]) -> io::Result { + self.0.seek(SeekFrom::Start(offset))?; + self.0.write(bytes) + } + + fn flush(&mut self) -> io::Result<()> { + self.0.sync_all() + } +} + +/// A section a document provider keeps, as its replica publishes to it. +pub(crate) struct ProviderRemote { + documents: Arc, + path: String, + /// The image the last stamp had to read whole, which the step's read takes. + read: Option>, +} + +impl ProviderRemote { + pub(crate) fn new(documents: &Arc, path: &str) -> Self { + Self { + documents: Arc::clone(documents), + path: path.to_owned(), + read: None, + } + } +} + +impl crate::Remote for ProviderRemote { + /// A torn or damaged image, as a stream mid-upload reads, is never a base to rebase on. + fn read(&mut self) -> io::Result> { + let image = match self.read.take() { + 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()) { + return Err(io::Error::new( + io::ErrorKind::WouldBlock, + "The section was read mid-write", + )); + } + Ok(image) + } + + fn stamp(&mut self) -> io::Result { + let (stamp, image) = self.documents.stamp(&self.path)?; + self.read = image; + Ok(stamp) + } + + fn publish(&mut self, transaction: &Transaction) -> std::result::Result<(), CommitError> { + self.read = None; + self.documents + .write(&self.path, |io| transaction.commit(io)) + } + + fn confirm(&mut self, base: &Stamp) -> std::result::Result<(), CommitError> { + self.read = None; + self.documents + .write(&self.path, |io| onestore::confirm(io, base)) + } +} + +/// A notebook folder a document provider keeps, its folders as last listed kept at `listed`. +pub(crate) struct Tree { + pub(crate) documents: Arc, + pub(crate) listed: PathBuf, +} + +/// The provider's folders, or as they were last listed while it can't be reached. +struct Source<'a> { + documents: &'a Documents, + kept: BTreeMap>, + listed: BTreeMap>, + /// Documents sharing a name with one listed, by folder. + duplicates: BTreeMap>, +} + +impl discover::Source for Source<'_> { + fn entries(&mut self, path: &str, limit: usize) -> io::Result> { + let entries = match self.documents.listing(path) { + Ok((entries, duplicates)) => { + self.duplicates.insert(path.to_owned(), duplicates); + entries + } + Err(error) if error.kind() == io::ErrorKind::NotConnected => { + self.kept.get(path).cloned().ok_or(error)? + } + Err(error) => return Err(error), + }; + if entries.len() > limit { + return Err(io::ErrorKind::FileTooLarge.into()); + } + self.listed.insert(path.to_owned(), entries.clone()); + Ok(entries) + } + + fn read(&mut self, path: &str, limit: usize) -> io::Result> { + self.documents.read(path, limit) + } +} + +fn unsupported() -> Error { + io::Error::new( + io::ErrorKind::Unsupported, + "Snowbound can't change this notebook's folders in another app yet", + ) + .into() +} + +impl Storage for Tree { + /// Of several documents with one name, the first by id stands for it and the rest are + /// unavailable, named with their ids. + fn discover( + &self, + cache: &mut discover::Cache, + limits: discover::Limits, + ) -> Result { + let kept = crate::fs::read(&self.listed) + .ok() + .and_then(|bytes| serde_json::from_slice(&bytes).ok()) + .unwrap_or_default(); + let mut source = Source { + documents: &self.documents, + kept, + listed: BTreeMap::new(), + duplicates: BTreeMap::new(), + }; + let mut catalog = cache.discover(&mut source, limits)?; + if let Ok(json) = serde_json::to_vec(&source.listed) { + let _ = crate::fs::write(&self.listed, json); + } + for (path, duplicates) in source.duplicates { + let Some(folder) = folder(&mut catalog, &path) else { + continue; + }; + for duplicate in duplicates { + folder.unavailable.push(discover::Unavailable { + path: discover::join(&path, &format!("{} ({})", duplicate.name, duplicate.id)), + group: duplicate.directory, + error: "Another document here has the same name".into(), + reason: discover::Reason::Denied, + }); + } + } + Ok(catalog) + } + + fn location(&self) -> String { + self.documents.location() + } + + fn entries(&self, folder: &str) -> io::Result> { + self.documents.entries(folder) + } + + fn exists(&self, path: &str) -> bool { + let (folder, name) = path.rsplit_once('/').unwrap_or(("", path)); + self.documents + .entries(folder) + .is_ok_and(|entries| entries.iter().any(|entry| entry.name == name)) + } + + fn stamp(&self, path: &str) -> io::Result { + Ok(self.documents.stamp(path)?.0) + } + + fn read(&self, path: &str) -> Result> { + Ok(self.documents.read(path, crate::MAX_FILE_BYTES)?) + } + + fn read_file(&self, path: &str, limit: usize) -> Result> { + Ok(self.documents.read(path, limit)?) + } + + fn create(&self, _: &str, _: &[u8]) -> Result<()> { + Err(unsupported()) + } + + fn create_directory(&self, _: &str) -> Result<()> { + Err(unsupported()) + } + + fn hide(&self, _: &str) -> Result<()> { + Err(unsupported()) + } + + fn rename(&self, from: &str, to: &str) -> Result<()> { + let folder = |path: &str| path.rsplit_once('/').map(|(folder, _)| folder.to_owned()); + if folder(from) != folder(to) { + return Err(unsupported()); + } + let name = to.rsplit('/').next().unwrap_or(to); + Ok(self.documents.rename(from, name)?) + } + + fn rename_root(&self, _: &str, _: &[String]) -> Result { + Err(unsupported()) + } + + fn replace(&self, _: &str, _: &str) -> Result<()> { + Err(unsupported()) + } + + fn delete(&self, _: &str) -> Result<()> { + Err(unsupported()) + } + + fn place(&self, path: &str, ancestor: [u8; 16], name: &str) -> Result<()> { + Ok(self.documents.write(path, |io| { + onestore::place(io, ancestor, name).map_err(|error| CommitError { + state: CommitState::Unknown, + error, + }) + })?) + } + + fn commit(&self, path: &str, transaction: &Transaction) -> Result<()> { + Ok(self.documents.write(path, |io| transaction.commit(io))?) + } + + fn confirm(&self, path: &str, base: &Stamp) -> std::result::Result<(), CommitError> { + self.documents.write(path, |io| onestore::confirm(io, base)) + } + + fn supersede(&self, _: &str, _: &Stamp, _: &str) -> Result<()> { + Err(unsupported()) + } +} + +fn folder<'a>(folder: &'a mut discover::Folder, path: &str) -> Option<&'a mut discover::Folder> { + if folder.path == path { + return Some(folder); + } + folder + .groups + .iter_mut() + .find_map(|group| self::folder(group, path)) +} + +#[cfg(all(test, unix))] +mod tests; diff --git a/crates/notebook/src/provider/tests.rs b/crates/notebook/src/provider/tests.rs new file mode 100644 index 0000000000000000000000000000000000000000..bcfa2b464f2bf40eb5832a90b3878e296299ce4c --- /dev/null +++ b/crates/notebook/src/provider/tests.rs @@ -0,0 +1,808 @@ +//! A fake document provider with real descriptors: seekable files, or pipes as the Android lab +//! measured stream-only providers (`tools/android/provider-lab`). + +use super::*; +use crate::{ + EditStatus, Replica, + session::{self, Background, Event, Notebook, SyncState, SyncStatus}, +}; +use onestore::{ + ExGuid, + op::{Edit, Op, PageOp}, + page::{Page, PageObject}, +}; +use std::{ + fs, + ops::Range, + os::fd::OwnedFd, + path::Path, + thread, + time::{Duration, Instant}, +}; + +const TOC: &str = "Open Notebook.onetoc2"; + +#[derive(Clone, Copy, PartialEq)] +enum Kind { + Seekable, + /// Pipes in every mode, `rw` included, with no metadata. + Stream, +} + +struct Node { + parent: String, + name: String, + directory: bool, +} + +struct State { + kind: Kind, + nodes: BTreeMap, + next: u32, + writable: bool, + offline: bool, + revoked: bool, + /// One device's writes reach the others only at `publish_staged`, as the lab's + /// upload-on-close provider showed. + delayed: bool, + /// 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>, + modes: Vec, +} + +struct Shared { + folder: tempfile::TempDir, + state: Mutex, +} + +/// One device's view of the provider. +#[derive(Clone)] +struct Fake { + store: Arc, + device: u32, +} + +impl Fake { + fn new(kind: Kind) -> Self { + let mut nodes = BTreeMap::new(); + nodes.insert( + "root".to_owned(), + Node { + parent: String::new(), + name: String::new(), + directory: true, + }, + ); + let state = State { + kind, + nodes, + next: 0, + writable: true, + offline: false, + revoked: false, + delayed: false, + swap: None, + modes: Vec::new(), + }; + Self { + store: Arc::new(Shared { + folder: tempfile::tempdir().unwrap(), + state: Mutex::new(state), + }), + device: 0, + } + } + + fn device(&self, device: u32) -> Self { + Self { + store: Arc::clone(&self.store), + device, + } + } + + fn documents(&self, root: &str) -> Arc { + Arc::new(Documents::new(self.clone(), root)) + } + + fn state(&self) -> MutexGuard<'_, State> { + lock(&self.store.state) + } + + fn file(&self, id: &str) -> PathBuf { + self.store.folder.path().join(id) + } + + fn staged(&self, id: &str) -> PathBuf { + self.store + .folder + .path() + .join(format!("{id}.{}", self.device)) + } + + /// The bytes this device reads. + fn visible(&self, id: &str) -> PathBuf { + let staged = self.staged(id); + if staged.exists() { + staged + } else { + self.file(id) + } + } + + fn add(&self, name: &str, bytes: &[u8]) -> String { + let mut state = self.state(); + state.next += 1; + let id = format!("doc-{:04}", state.next); + state.nodes.insert( + id.clone(), + Node { + parent: "root".into(), + name: name.into(), + directory: false, + }, + ); + fs::write(self.file(&id), bytes).unwrap(); + id + } + + /// The first document named `name` at the root, as stored. + fn id(&self, name: &str) -> String { + let state = self.state(); + let (id, _) = state + .nodes + .iter() + .find(|(_, node)| node.parent == "root" && node.name == name) + .unwrap(); + id.clone() + } + + fn bytes(&self, name: &str) -> Vec { + fs::read(self.file(&self.id(name))).unwrap() + } + + fn set(&self, change: impl FnOnce(&mut State)) { + change(&mut self.state()); + } + + fn modes(&self) -> Vec { + self.state().modes.clone() + } + + /// Makes every device's writes the document, in device order, as an explicit provider + /// publication does. + fn publish_staged(&self) { + let ids: Vec = self.state().nodes.keys().cloned().collect(); + for id in ids { + for device in 0..4 { + let staged = self.device(device).staged(&id); + if staged.exists() { + fs::rename(staged, self.file(&id)).unwrap(); + } + } + } + } + + /// Gives the document a new id, as another app renaming it does; the old one is gone. + fn reidentify(&self, name: &str) -> String { + let old = self.id(name); + let mut state = self.state(); + state.next += 1; + let new = format!("doc-{:04}", state.next); + let node = state.nodes.remove(&old).unwrap(); + state.nodes.insert(new.clone(), node); + fs::rename(self.file(&old), self.file(&new)).unwrap(); + new + } + + fn reachable(&self) -> io::Result> { + let state = self.state(); + if state.offline { + return Err(io::Error::new(io::ErrorKind::NotConnected, "Offline")); + } + if state.revoked { + return Err(io::Error::new(io::ErrorKind::PermissionDenied, "Revoked")); + } + Ok(state) + } + + fn describe(&self, state: &State, id: &str) -> io::Result { + let node = state.nodes.get(id).ok_or(io::ErrorKind::NotFound)?; + let metadata = fs::metadata(self.visible(id)).ok(); + let known = |value: u64| (state.kind == Kind::Seekable).then_some(value); + Ok(Document { + id: id.to_owned(), + name: node.name.clone(), + directory: node.directory, + size: metadata.as_ref().and_then(|metadata| known(metadata.len())), + modified: metadata + .and_then(|metadata| metadata.modified().ok()) + .and_then(|at| at.duration_since(std::time::UNIX_EPOCH).ok()) + .and_then(|since| known(since.as_nanos() as u64)), + writable: state.writable, + }) + } +} + +impl Provider for Fake { + fn document(&self, id: &str) -> io::Result { + let state = self.reachable()?; + self.describe(&state, id) + } + + fn children(&self, id: &str) -> io::Result> { + let state = self.reachable()?; + if !state + .nodes + .get(id) + .ok_or(io::ErrorKind::NotFound)? + .directory + { + return Err(io::ErrorKind::NotADirectory.into()); + } + let children = state.nodes.iter().filter(|(_, node)| node.parent == id); + children.map(|(id, _)| self.describe(&state, id)).collect() + } + + fn open(&self, id: &str, mode: &str) -> io::Result { + let mut state = self.reachable()?; + if !state.nodes.contains_key(id) { + return Err(io::ErrorKind::NotFound.into()); + } + state.modes.push(mode.to_owned()); + if mode != "r" && !state.writable { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "Read-only grant", + )); + } + if mode == "rw" + && let Some(twin) = state.swap.take() + { + let mut node = state.nodes.remove(id).unwrap(); + let name = std::mem::replace(&mut node.name, "Moved.one".into()); + let parent = node.parent.clone(); + let (moved, taken) = (format!("{id}-moved"), format!("{id}-taken")); + state.nodes.insert(moved.clone(), node); + let directory = false; + state.nodes.insert( + taken.clone(), + Node { + parent, + name, + directory, + }, + ); + fs::rename(self.file(id), self.file(&moved))?; + fs::write(self.file(&taken), twin)?; + return Err(io::ErrorKind::NotFound.into()); + } + match (state.kind, mode) { + (Kind::Seekable, "r") => File::open(self.visible(id)), + (Kind::Seekable, _) => { + let mut file = self.file(id); + if state.delayed { + let staged = self.staged(id); + if !staged.exists() { + fs::copy(&file, &staged)?; + } + file = staged; + } + File::options() + .read(true) + .write(true) + .truncate(mode == "wt") + .open(file) + } + (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))), + } + } + + fn close(&self, _: File) -> io::Result<()> { + Ok(()) + } + + fn rename(&self, id: &str, name: &str) -> io::Result { + let mut state = self.reachable()?; + let mut node = state.nodes.remove(id).ok_or(io::ErrorKind::NotFound)?; + node.name = name.to_owned(); + state.next += 1; + let new = format!("doc-{:04}", state.next); + state.nodes.insert(new.clone(), node); + fs::rename(self.file(id), self.file(&new))?; + Ok(new) + } +} + +/// 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 file = |image: &[u8]| Store::parse(image).unwrap().header.file_id; + let toc = onestore::create_table_of_contents( + TOC, + &[("Notes.one", file(¬es)), ("Ideas.one", file(&ideas))], + ) + .unwrap(); + onestore::place_image(&mut notes, Some((file(&toc), "Notes.one"))).unwrap(); + onestore::place_image(&mut ideas, Some((file(&toc), "Ideas.one"))).unwrap(); + fake.add(TOC, &toc); + fake.add("Notes.one", ¬es); + fake.add("Ideas.one", &ideas); +} + +/// Opens a section of `notebook` as a host does: its replica, made from the file only where +/// there is none. +fn open(notebook: &Notebook, documents: &Arc, path: &str) -> session::Section { + let cache = notebook.replica_path(path).unwrap(); + fs::create_dir_all(cache.parent().unwrap()).unwrap(); + let replica = Replica::open_or_create(&cache, None, || notebook.read_section(path)).unwrap(); + session::Section::resume_provider(path.into(), replica, Arc::clone(documents), false, || {}) + .unwrap() +} + +fn paragraph(page: &Page) -> (ExGuid, String) { + page.objects + .iter() + .find_map(|object| match object { + PageObject::Outline(outline) => outline.paragraphs.iter().find_map(|paragraph| { + let text = paragraph.text()?; + Some((text.id, text.text.text().to_owned())) + }), + _ => None, + }) + .unwrap() +} + +fn first_page(section: &session::Section) -> (ExGuid, Page) { + let space = section.pages().unwrap()[0].0; + (space, section.page(space).unwrap()) +} + +fn text(section: &session::Section) -> String { + paragraph(&first_page(section).1).1 +} + +/// Replaces `range` of the first page's text as the editor does, returning the edit's id. +fn typed(section: &session::Section, range: Range, with: &str) -> u64 { + let (space, page) = first_page(section); + let op = PageOp::Text { + text: paragraph(&page).0, + range, + with: with.into(), + }; + let edit = Edit { + at: crate::now(), + ops: vec![Op::Page { space, op }], + }; + section.replica().apply("Editor", edit).unwrap() +} + +/// The first page's text in `image`, which must cold-open as a section. +fn stored(image: Vec) -> String { + let arena = onestore::Arena::default(); + let mut section = onestore::Section::open(&arena, image).unwrap(); + let space = section.pages().unwrap()[0].0; + paragraph(§ion.page(space).unwrap()).1 +} + +fn until(what: &str, mut done: impl FnMut() -> bool) { + let deadline = Instant::now() + Duration::from_secs(20); + while !done() { + assert!(Instant::now() < deadline, "{what}"); + thread::sleep(Duration::from_millis(20)); + } +} + +fn published(section: &session::Section, id: u64) { + until("the edit was published", || { + section.wake(); + matches!( + section.status(id).unwrap(), + Some(EditStatus::Published { .. }) + ) + }); +} + +fn status(section: &session::Section, accept: impl Fn(&SyncStatus) -> bool) -> SyncStatus { + let mut last = None; + until("the status came", || { + section.wake(); + let status = section.sync_status().unwrap(); + let accepted = accept(&status); + last = Some(status); + accepted + }); + last.unwrap() +} + +#[test] +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(); + assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]); + let section = open(¬ebook, &documents, "Notes.one"); + let before = fake.bytes("Notes.one").len(); + let id = typed(§ion, 0..0, "Hello "); + published(§ion, id); + let after = fake.bytes("Notes.one"); + assert!(after.len() > before, "the revision was appended"); + assert_eq!(stored(after), "Hello Original text"); + 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 = open(&other, &elsewhere, "Notes.one"); + let id = typed(&other, 0..0, "World "); + published(&other, id); + until("the change reached the first device", || { + section.wake(); + text(§ion) == "World Hello Original text" + }); + assert!( + section + .events() + .iter() + .any(|event| matches!(event, Event::Changed(_))) + ); + assert_eq!(stored(fake.bytes("Notes.one")), "World Hello Original text"); + section.close().unwrap(); + other.close().unwrap(); +} + +#[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)); + } + 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] +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 section = open(¬ebook, &documents, "Notes.one"); + let before = fake.bytes("Notes.one"); + typed(§ion, 0..0, "Mine "); + let status = status(§ion, |status| status.error.is_some()); + assert_eq!(status.state(), SyncState::ReadOnly); + assert_eq!( + status.error.unwrap().kind(), + io::ErrorKind::ReadOnlyFilesystem + ); + assert_eq!(status.queued, 1); + assert_eq!(fake.bytes("Notes.one"), before); + assert!(fake.modes().iter().all(|mode| mode == "r")); + section.close().unwrap(); +} + +#[test] +fn a_write_another_device_sees_late_rebases_it_once_visible() { + let fake = Fake::new(Kind::Seekable); + 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, reading) = ( + open(&writing, &writer, "Notes.one"), + open(&reading, &reader, "Notes.one"), + ); + let id = typed(&writing, 0..0, "Early "); + published(&writing, id); + // The writer's close succeeded; the other device still reads the old bytes. + let old = reader.read("Notes.one", crate::MAX_FILE_BYTES).unwrap(); + assert_eq!(stored(old), "Original text"); + reading.wake(); + thread::sleep(Duration::from_millis(200)); + assert_eq!(text(&reading), "Original text"); + reading.set_offline(true); + let id = typed(&reading, 13..13, " later"); + fake.publish_staged(); + reading.set_offline(false); + published(&reading, id); + assert_eq!(text(&reading), "Early Original text later"); + fake.publish_staged(); + assert_eq!(stored(fake.bytes("Notes.one")), "Early Original text later"); + writing.close().unwrap(); + reading.close().unwrap(); +} + +#[test] +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 background = Background::provider(Arc::clone(&documents), || {}).unwrap(); + background.watch(notebook.replicas()); + until("every section has its offline copy", || { + ["Notes.one", "Ideas.one"] + .iter() + .all(|path| notebook.replica_path(path).unwrap().exists()) + }); + background.stop(); + let section = open(¬ebook, &documents, "Ideas.one"); + fake.set(|state| state.offline = true); + let id = typed(§ion, 0..0, "Offline "); + status(§ion, |status| { + status.state() == SyncState::NotConnected && status.queued == 1 + }); + section.close().unwrap(); + drop(notebook); + + // A new process: the provider fails throughout. + let documents = fake.documents("root"); + let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]); + let notes = open(¬ebook, &documents, "Notes.one"); + assert_eq!(text(¬es), "Original text"); + let section = open(¬ebook, &documents, "Ideas.one"); + assert_eq!(text(§ion), "Offline First idea"); + assert_eq!(section.pending().unwrap()[0].id, id); + status(§ion, |status| status.state() == SyncState::NotConnected); + fake.set(|state| state.offline = false); + published(§ion, id); + assert_eq!(stored(fake.bytes("Ideas.one")), "Offline First idea"); + notes.close().unwrap(); + section.close().unwrap(); +} + +#[test] +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 replica = notebook.replica_path("Notes.one").unwrap(); + let section = open(¬ebook, &documents, "Notes.one"); + fake.set(|state| state.revoked = true); + let id = typed(§ion, 0..0, "Kept "); + let status = status(§ion, |status| status.error.is_some()); + assert_eq!( + status.error.unwrap().kind(), + io::ErrorKind::PermissionDenied + ); + section.close().unwrap(); + drop(notebook); + let refused = Notebook::open_provider(fake.documents("root"), cache.path()).err(); + assert!( + matches!( + &refused, + Some(Error::Discovery(discover::Error::Io { error, .. })) + if error.kind() == io::ErrorKind::PermissionDenied + ), + "{refused:?}" + ); + assert_eq!( + Replica::open(&replica).unwrap().pending().unwrap()[0].id, + id + ); + + fake.set(|state| state.revoked = false); + let documents = fake.documents("root"); + let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap(); + let section = open(¬ebook, &documents, "Notes.one"); + published(§ion, id); + assert_eq!(stored(fake.bytes("Notes.one")), "Kept Original text"); + section.close().unwrap(); +} + +#[test] +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 section = open(¬ebook, &documents, "Notes.one"); + let id = typed(§ion, 0..0, "One "); + published(§ion, id); + fake.reidentify("Notes.one"); + let id = typed(§ion, 0..0, "Two "); + published(§ion, id); + assert_eq!(stored(fake.bytes("Notes.one")), "Two One Original text"); + section.close().unwrap(); + + let old = fake.id("Notes.one"); + assert_eq!( + notebook.rename("Notes.one", "Journal").unwrap(), + "Journal.one" + ); + assert!(matches!( + fake.document(&old).map_err(|error| error.kind()), + Err(io::ErrorKind::NotFound) + )); + assert_eq!(notebook.catalog().order, ["Journal.one", "Ideas.one"]); + let identity = notebook.catalog().sections[0].file_id; + assert!(session::lists(&fake.bytes(TOC), identity).unwrap()); + let section = open(¬ebook, &documents, "Journal.one"); + let id = typed(§ion, 0..0, "Three "); + published(§ion, id); + assert_eq!( + stored(fake.bytes("Journal.one")), + "Three Two One Original text" + ); + section.close().unwrap(); +} + +#[test] +fn of_documents_sharing_a_name_one_lists_and_the_rest_are_unavailable() { + let fake = Fake::new(Kind::Seekable); + notebook(&fake); + let twin = fake.add( + "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(); + assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]); + let unavailable = ¬ebook.catalog().unavailable; + assert_eq!(unavailable.len(), 1); + assert_eq!(unavailable[0].path, format!("Notes.one ({twin})")); + let section = open(¬ebook, &documents, "Notes.one"); + assert_eq!(text(§ion), "Original text"); + section.close().unwrap(); +} + +#[test] +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 section = open(¬ebook, &documents, "Notes.one"); + let id = typed(§ion, 0..0, "Checked "); + published(§ion, id); + let original = fake.bytes("Notes.one"); + let twin = onestore::create_section("Notes.one", "A twin", "Author").unwrap(); + fake.set(|state| state.swap = Some(twin.clone())); + typed(§ion, 0..0, "Refused "); + until("the write was refused", || { + section.wake(); + fake.state().swap.is_none() && section.sync_status().unwrap().error.is_some() + }); + for _ in 0..3 { + section.wake(); + thread::sleep(Duration::from_millis(100)); + } + assert_eq!(fake.bytes("Notes.one"), twin); + assert_eq!(fake.bytes("Moved.one"), original); + // Another section under the name stops the worker, as it does for a file. + assert!(section.close().is_err()); +} + +#[test] +fn place_and_confirm_write_a_seekable_document_in_place() { + let fake = Fake::new(Kind::Seekable); + notebook(&fake); + let listed = tempfile::tempdir().unwrap(); + let tree = Tree { + documents: fake.documents("root"), + listed: listed.path().join("entries.json"), + }; + let before = fake.bytes("Ideas.one"); + let stamp = Stamp::of(&before).unwrap(); + tree.confirm("Ideas.one", &stamp).unwrap(); + let confirmed = fake.bytes("Ideas.one"); + assert_eq!(confirmed.len(), before.len()); + assert_ne!(confirmed[212..252], before[212..252]); + assert_eq!(stored(confirmed.clone()), "First idea"); + let stale = tree.confirm("Ideas.one", &stamp).unwrap_err(); + assert_eq!(stale.state, CommitState::NotCommitted); + tree.place("Ideas.one", [7; 16], "Renamed.one").unwrap(); + let mut placed = confirmed; + onestore::place_image(&mut placed, Some(([7; 16], "Renamed.one"))).unwrap(); + assert_eq!(fake.bytes("Ideas.one"), placed); + assert!(fake.modes().iter().all(|mode| mode == "r" || mode == "rw")); +} + +#[test] +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, remote) = ( + open(&local, &mine, "Notes.one"), + open(&remote, &theirs, "Notes.one"), + ); + local.set_offline(true); + typed(&local, 0..8, "Local"); + let id = typed(&remote, 0..8, "Native"); + published(&remote, id); + local.set_offline(false); + until("the conflict page was published", || { + local.wake(); + !local.conflicts().unwrap().is_empty() && local.pending().unwrap().is_empty() + }); + assert_eq!(text(&local), "Native text"); + let conflicts = local.conflicts().unwrap(); + let kept = local.page(conflicts[0].1[0].space).unwrap(); + assert_eq!(paragraph(&kept).1, "Local text"); + let arena = onestore::Arena::default(); + let mut stored = onestore::Section::open(&arena, fake.bytes("Notes.one")).unwrap(); + assert_eq!(stored.conflicts().unwrap().len(), 1); + local.close().unwrap(); + remote.close().unwrap(); +} + +#[test] +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 section = open(&fake); + let id = typed(§ion, 0..0, "Not "); + published(§ion, id); + section.close().unwrap(); + fake.set(|state| state.offline = true); + let section = open(&fake); + assert_eq!(section.file(), Path::new(&lone)); + assert_eq!(text(§ion), "Not Alone"); + let id = typed(§ion, 9..9, " now"); + status(§ion, |status| status.state() == SyncState::NotConnected); + fake.set(|state| state.offline = false); + published(§ion, id); + assert_eq!(stored(fake.bytes("Lone.one")), "Not Alone now"); + section.close().unwrap(); +} diff --git a/crates/notebook/src/session.rs b/crates/notebook/src/session.rs index 93cc3a8f2e5bd4d95395b7d59a86cbae5ffe36c6..addc69691e51303f449ad567f0c69f79ae6122ad 100644 --- a/crates/notebook/src/session.rs +++ b/crates/notebook/src/session.rs @@ -514,6 +514,21 @@ 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"); + Self::with( + Box::new(crate::provider::Tree { documents, listed }), + None, + cache, + ) + } + /// The storage the notebook's files are in, for a Live Share host to serve. pub fn into_storage(self) -> Box { self.storage @@ -1844,14 +1859,21 @@ impl Notebook { if fs::metadata(&cache).is_ok() { let (replica, remote, mut connect) = (Replica::open(&cache)?, file.clone(), connect); - return Section::start(file, replica, move || connect(&remote), offline, notify); + return Section::start( + file, + replica, + move || connect(&remote), + CHECK, + offline, + notify, + ); } } let replica = |identity: &[u8; 16], _: &[u8]| { fs::create_dir_all(&replicas)?; Ok(replica_file(&replicas, identity)) }; - Section::open_in(file, key, replica, connect, offline, notify) + Section::open_in(file, key, replica, connect, CHECK, offline, notify) } } @@ -2110,6 +2132,22 @@ fn remove_replica(replica: &Path) -> io::Result<()> { Ok(()) } +/// How often an open section's file is checked where nothing watches it. +const CHECK: Duration = Duration::from_secs(2); + +/// Names a lone section's replica in the folder of `location` by its identity, taking one a +/// moved file left behind (`location::claim`). +fn claimed<'a>( + cache: &'a Path, + location: &'a str, +) -> impl FnOnce(&[u8; 16], &[u8]) -> Result + 'a { + move |identity, source| { + let folder = + crate::location::claim(cache, location, &[*identity], |_| Stamp::of(source).ok())?; + Ok(replica_file(&folder, identity)) + } +} + /// The replica in `cache` of the section `identity` names. pub(crate) fn replica_file(cache: &Path, identity: &[u8; 16]) -> PathBuf { let name: String = identity.iter().map(|byte| format!("{byte:02x}")).collect(); @@ -2264,13 +2302,8 @@ impl Section { ) -> Result { let file = fs::canonicalize(file)?; let location = file.to_string_lossy().into_owned(); - let replicas = |identity: &[u8; 16], source: &[u8]| { - let folder = crate::location::claim(cache.as_ref(), &location, &[*identity], |_| { - Stamp::of(source).ok() - })?; - Ok(replica_file(&folder, identity)) - }; - Self::open_in(file, None, replicas, connect, offline, notify) + let replicas = claimed(cache.as_ref(), &location); + Self::open_in(file, None, replicas, connect, CHECK, offline, notify) } /// Opens the canonical section `file`, a protected one under `key`, through the replica @@ -2280,6 +2313,7 @@ impl Section { key: Option<&protected::Key>, replica: impl FnOnce(&[u8; 16], &[u8]) -> Result, mut connect: impl FnMut(&Path) -> io::Result + Send + 'static, + interval: Duration, offline: bool, notify: impl Fn() + Send + 'static, ) -> Result { @@ -2289,7 +2323,14 @@ impl Section { let cache = replica(&identity.guid, &source)?; let replica = Replica::open_or_create(&cache, key, || Ok(source))?; let remote = file.clone(); - Self::start(file, replica, move || connect(&remote), offline, notify) + Self::start( + file, + replica, + move || connect(&remote), + interval, + offline, + notify, + ) } /// Resumes an owned local replica without reading the publication target. @@ -2306,6 +2347,7 @@ impl Section { file, replica, move || Ok(FileRemote(remote.clone())), + CHECK, offline, notify, ) @@ -2326,6 +2368,7 @@ impl Section { PathBuf::from(&path), replica, move || Ok(crate::SmbRemote::new(connect()?, path.clone(), limit)), + CHECK, offline, notify, ) @@ -2344,15 +2387,84 @@ impl Section { PathBuf::from(&path), replica, move || Ok(crate::live::share::HostedRemote::new(&guest, &path)), + CHECK, offline, notify, ) } + /// Resumes a replica against the section at catalog `path` of a notebook a document + /// provider keeps (`Notebook::open_provider`). Its file is checked every + /// `Background::UNWATCHED`, and when `wake` passes on what the host hears of changes. + pub fn resume_provider( + path: String, + replica: Replica, + documents: Arc, + offline: bool, + notify: impl Fn() + Send + 'static, + ) -> Result { + Self::start( + PathBuf::from(&path), + replica, + move || Ok(crate::provider::ProviderRemote::new(&documents, &path)), + Background::UNWATCHED, + offline, + notify, + ) + } + + /// Opens the lone section that is the root of `documents` as `open` opens a file, checked + /// as `resume_provider`'s is; while the provider can't be reached, from the only replica + /// 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 file = PathBuf::from(&documents.root); + let remote = move || Ok(crate::provider::ProviderRemote::new(&documents, "")); + let notify = Arc::new(notify); + let (connect, woken) = (remote.clone(), Arc::clone(¬ify)); + let replicas = claimed(cache, &location); + let opened = Self::open_in( + file.clone(), + None, + replicas, + move |_| connect(), + Background::UNWATCHED, + offline, + move || woken(), + ); + match opened { + Err(Error::Io(error)) if error.kind() == io::ErrorKind::NotConnected => { + let mut replicas = fs::read_dir(crate::location::folder(cache, &location)) + .into_iter() + .flatten() + .filter_map(|entry| Some(entry.ok()?.path())) + .filter(|path| path.extension() == Some("sqlite".as_ref())); + match (replicas.next(), replicas.next()) { + (Some(only), None) => Self::start( + file, + Replica::open(only)?, + remote, + Background::UNWATCHED, + offline, + move || notify(), + ), + _ => Err(Error::Io(error)), + } + } + opened => opened, + } + } + fn start( file: PathBuf, replica: Replica, connect: impl FnMut() -> io::Result + Send + 'static, + interval: Duration, offline: bool, notify: impl Fn() + Send + 'static, ) -> Result { @@ -2374,20 +2486,21 @@ impl Section { notify, observed, }; - section.worker = Some(section.start_worker(connect, offline, None)?); + section.worker = Some(section.start_worker(connect, interval, offline, None)?); Ok(section) } fn start_worker( &self, connect: impl FnMut() -> io::Result + Send + 'static, + interval: Duration, offline: bool, replacing: Option<&SyncWorker>, ) -> Result { let (sender, notify) = (self.sender.clone(), Arc::clone(&self.notify)); let observed = Arc::clone(&self.observed); Ok(self.replica.start_sync_replacing( - Duration::from_secs(2), + interval, offline, connect, move |result| { @@ -2455,7 +2568,7 @@ impl Section { } self.observed = Arc::new(Mutex::new(None)); self.worker = - Some(self.start_worker(move || connect(&target), false, previous.as_ref())?); + Some(self.start_worker(move || connect(&target), CHECK, false, previous.as_ref())?); self.file = file; Ok(()) } diff --git a/crates/notebook/src/session/tests.rs b/crates/notebook/src/session/tests.rs index 5e648b2a75cc8bb2e058e9cf41f85714566620c4..db6d36940f8a936e4f307cff89c3644ac2278604 100644 --- a/crates/notebook/src/session/tests.rs +++ b/crates/notebook/src/session/tests.rs @@ -568,6 +568,7 @@ fn initially_offline_workers_keep_the_original_queue_until_explicitly_woken() { calls: Arc::clone(&counter), }) }, + CHECK, true, move || { let _ = notify.send(()); -- 2.54.0