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(());