authorclover caruso <git@paperclover.net> 2026-10-15 09:16:46-07:00
committerclover caruso <git@paperclover.net> 2026-10-15 10:37:57-07:00
logffaff48d67462562d069edde11886b41bb411e8c
treeffa11fa950a5b60dbc618c7024501e0677fabf00
parent08dbdad9e4fb203c874a7035cc77f5e4db5bc7b8
signature Signed by SSH key SHA256:52mNGHRsVFBDED9IAX5pe+LRWUefqTbxEReunq21QvU (~clover)

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

8 files changed, 1459 insertions(+), 16 deletions(-)

crates/notebook/README.md+27
......@@ -545,6 +545,33 @@ Device and simulator builds link for iOS. Native acceptance uses disposable
545545OneNote 2010 clients and Samba; it does not establish on-device execution or
546546physical power-loss durability.
547547
548## Document providers
549
550`provider` opens notebooks a document provider keeps, as Android's Storage Access Framework
551reaches them. The host implements `provider::Provider` (a document's description, a folder's
552children, `open` in `ContentResolver`'s `r` or `rw` as an owned `File`, `close` with the provider's
553verdict, `rename` returning the new id) and reports an unreachable provider as `NotConnected`,
554a revoked grant as `PermissionDenied` and a gone document as `NotFound`.
555`provider::Documents::new(provider, root)` holds one root: a picked folder opens
556with `Notebook::open_provider(documents, cache)`, its sections with `Section::resume_provider`
557and the rest with `Background::provider`, as a Live Share guest's do; a picked document opens
558as a lone section with `Section::open_provider`. Catalog paths resolve to ids by listing their
559folders, again once when an id has gone, as a rename leaves it; of documents sharing a name the
560first by id lists and the rest are `unavailable`, named with their ids.
561
562A publication resolves the document's id once, from the last listing, and writes nothing to
563a document that took its name since. Without a write grant a document is only read, and
564publication is refused as `ReadOnlyFilesystem`; where its `r` descriptor seeks, `rw` opens it
565and a commit checks the stamp and appends in place exactly as on a disk. A document that only
566streams is read whole for each stamp and is read-only too, its edits waiting, until lost
567publications merge back. No provider offers a lock or compare-and-swap: the check sees only
568this device's view of the document. An image that does not parse is never a base. An open
569section checks its file every `Background::UNWATCHED`, the host's change reports waking it
570between. While the provider can't be reached the notebook opens as its folders were last
571listed and sections from their replicas; a revoked grant keeps them, and the notebook does not
572open until access returns. Creating, deleting and moving documents and renaming the folder are
573`Unsupported`; renaming within a folder is not.
574
548575## Live presence and Live Share (feature `live`)
549576
550577`live::Live::start(hello, room, reach, relay, events)` listens on a TCP port and, with a
crates/notebook/src/background.rs+19
......@@ -392,6 +392,25 @@ impl Background {
392392 Ok(background)
393393 }
394394
395 /// Keeps the sections of a notebook a document provider keeps in sync while they are not
396 /// open, with an offline copy of each. Nothing reports the folder's changes; the host
397 /// passes what it hears of them to `touched`. Paths are catalog paths.
398 pub fn provider(
399 documents: Arc<crate::provider::Documents>,
400 notify: impl Fn() + Send + 'static,
401 ) -> Result<Self> {
402 Self::start(
403 true,
404 move |_| {
405 let (bound, listing) = (Arc::clone(&documents), Arc::clone(&documents));
406 let bind = move |path: &str| crate::provider::ProviderRemote::new(&bound, path);
407 let list = move |folder: String| std::future::ready(listing.entries(&folder));
408 Ok(((bind, list), false))
409 },
410 notify,
411 )
412 }
413
395414 /// Checks what a report names after `settle` rather than a second, where reports come
396415 /// one to a commit, as a Live Share host sends them.
397416 pub fn set_settle(&self, settle: Duration) {
crates/notebook/src/discover.rs+3-3
......@@ -125,7 +125,7 @@ pub struct TocReference {
125125 pub order: u32,
126126}
127127
128#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
128#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
129129pub enum EntryKind {
130130 File,
131131 Directory,
......@@ -135,7 +135,7 @@ pub enum EntryKind {
135135 Evicted,
136136}
137137
138#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
138#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
139139pub struct Entry {
140140 pub name: String,
141141 pub kind: EntryKind,
......@@ -841,7 +841,7 @@ fn component(name: &str) -> bool {
841841 !name.is_empty() && name != "." && name != ".." && !name.contains(['/', '\\', '\0'])
842842}
843843
844fn join(parent: &str, name: &str) -> String {
844pub(crate) fn join(parent: &str, name: &str) -> String {
845845 if parent.is_empty() {
846846 name.into()
847847 } else {
crates/notebook/src/lib.rs+1
......@@ -32,6 +32,7 @@ pub mod location;
3232mod merge;
3333mod migrate;
3434pub mod package;
35pub mod provider;
3536mod queue;
3637mod recovery;
3738mod resolve;
crates/notebook/src/provider.rs created+474
......@@ -0,0 +1,474 @@
1//! Notebooks a document provider keeps, as Android's Storage Access Framework reaches them:
2//! documents named by opaque ids, listed by folder and opened as descriptors, with no locks,
3//! no rename over another document and, on many providers, no seeking. A picked folder is a
4//! notebook root (`session::Notebook::open_provider`), a picked document a lone section
5//! (`session::Section::open_provider`); replicas and the catalog's listing stay in the cache.
6
7use crate::{Error, Result, discover, session::Storage};
8use onestore::{CommitError, CommitIo, CommitState, Stamp, Store, Transaction};
9use std::{
10 collections::BTreeMap,
11 fs::File,
12 io::{self, Read, Seek, SeekFrom, Write},
13 path::PathBuf,
14 sync::{Arc, Mutex, MutexGuard, PoisonError},
15};
16
17/// A document as its provider describes it.
18#[derive(Clone, Debug)]
19pub struct Document {
20 pub id: String,
21 pub name: String,
22 pub directory: bool,
23 /// What the provider lists, where it lists anything.
24 pub size: Option<u64>,
25 pub modified: Option<u64>,
26 /// The grant allows writing and the provider supports it.
27 pub writable: bool,
28}
29
30/// The host's document provider. Failures take the shapes a provider gives them: unreachable
31/// as `NotConnected`, a revoked grant as `PermissionDenied`, and a document gone (or renamed,
32/// which changes its id) as `NotFound`.
33pub trait Provider: Send + Sync {
34 fn document(&self, id: &str) -> io::Result<Document>;
35 fn children(&self, id: &str) -> io::Result<Vec<Document>>;
36 /// Opens a document in a `ContentResolver` mode, `r` or `rw`.
37 fn open(&self, id: &str, mode: &str) -> io::Result<File>;
38 /// Closes a file `open` gave, with the provider's verdict on what was written to it.
39 fn close(&self, file: File) -> io::Result<()>;
40 /// Renames a document, returning its new id.
41 fn rename(&self, id: &str, name: &str) -> io::Result<String>;
42}
43
44/// The documents under one root a provider keeps: a notebook's folder or a lone section.
45/// Paths are catalog paths, `/`-separated below the root.
46pub struct Documents {
47 provider: Box<dyn Provider>,
48 pub(crate) root: String,
49 /// Each path's document as last listed; a stale one is listed again.
50 ids: Mutex<BTreeMap<String, String>>,
51}
52
53impl Documents {
54 pub fn new(provider: impl Provider + 'static, root: impl Into<String>) -> Self {
55 Self {
56 provider: Box::new(provider),
57 root: root.into(),
58 ids: Mutex::default(),
59 }
60 }
61
62 pub(crate) fn location(&self) -> String {
63 format!("provider:{}", self.root)
64 }
65
66 fn id(&self, path: &str) -> io::Result<String> {
67 if path.is_empty() {
68 return Ok(self.root.clone());
69 }
70 if let Some(id) = lock(&self.ids).get(path) {
71 return Ok(id.clone());
72 }
73 self.listing(path.rsplit_once('/').map_or("", |(folder, _)| folder))?;
74 lock(&self.ids)
75 .get(path)
76 .cloned()
77 .ok_or_else(|| io::ErrorKind::NotFound.into())
78 }
79
80 /// Reads through the document at `path`, listing its folders again once if the id they
81 /// last gave is gone.
82 fn with_id<T>(&self, path: &str, read: impl Fn(&str) -> io::Result<T>) -> io::Result<T> {
83 match read(&self.id(path)?) {
84 Err(error) if error.kind() == io::ErrorKind::NotFound && !path.is_empty() => {
85 lock(&self.ids).clear();
86 read(&self.id(path)?)
87 }
88 result => result,
89 }
90 }
91
92 /// A folder's entries, and the documents sharing a name with one listed before them by id.
93 fn listing(&self, folder: &str) -> io::Result<(Vec<discover::Entry>, Vec<Document>)> {
94 let mut children = self.with_id(folder, |id| self.provider.children(id))?;
95 children.sort_by(|a, b| (&a.name, &a.id).cmp(&(&b.name, &b.id)));
96 let (mut entries, mut duplicates) = (Vec::<discover::Entry>::new(), Vec::new());
97 let mut ids = lock(&self.ids);
98 for child in children {
99 if entries.last().is_some_and(|last| last.name == child.name) {
100 duplicates.push(child);
101 continue;
102 }
103 ids.insert(discover::join(folder, &child.name), child.id.clone());
104 entries.push(discover::Entry {
105 name: child.name,
106 kind: if child.directory {
107 discover::EntryKind::Directory
108 } else {
109 discover::EntryKind::File
110 },
111 listed: discover::Listed {
112 size: child.size.unwrap_or_default(),
113 // Unknown, it never matches an earlier listing, so the file is read again.
114 modified: child
115 .modified
116 .filter(|_| child.size.is_some())
117 .unwrap_or_else(crate::now),
118 },
119 });
120 }
121 Ok((entries, duplicates))
122 }
123
124 pub(crate) fn entries(&self, folder: &str) -> io::Result<Vec<discover::Entry>> {
125 Ok(self.listing(folder)?.0)
126 }
127
128 /// Opens the document `id`, hands it to `operation` and closes it, failing if either fails.
129 fn opened<T>(
130 &self,
131 id: &str,
132 mode: &str,
133 operation: impl FnOnce(&mut File) -> io::Result<T>,
134 ) -> io::Result<T> {
135 let mut file = self.provider.open(id, mode)?;
136 let result = operation(&mut file);
137 let closed = self.provider.close(file);
138 let value = result?;
139 closed?;
140 Ok(value)
141 }
142
143 fn read(&self, path: &str, limit: usize) -> io::Result<Vec<u8>> {
144 self.with_id(path, |id| self.opened(id, "r", |file| whole(file, limit)))
145 }
146
147 /// The document's stamp, with its image where a stream had to be read whole for it.
148 fn stamp(&self, path: &str) -> io::Result<(Stamp, Option<Vec<u8>>)> {
149 self.with_id(path, |id| {
150 self.opened(id, "r", |file| {
151 if file.stream_position().is_err() {
152 let image = whole(file, crate::MAX_FILE_BYTES)?;
153 let stamp = Stamp::of(&image)
154 .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?;
155 return Ok((stamp, Some(image)));
156 }
157 let mut header = [0; 1024];
158 file.read_exact(&mut header)?;
159 let length = file.seek(SeekFrom::End(0))?;
160 Ok((Stamp { header, length }, None))
161 })
162 })
163 }
164
165 /// Writes the document at `path` in place through `operation`, provided its grant allows
166 /// writing and it opens seekable. The id is the one the last listing gave, never listed
167 /// again, so that a document taking the name meanwhile is not the one written.
168 fn write(
169 &self,
170 path: &str,
171 operation: impl FnOnce(&mut Seekable) -> std::result::Result<(), CommitError>,
172 ) -> std::result::Result<(), CommitError> {
173 let refused = |error| CommitError {
174 state: CommitState::NotCommitted,
175 error,
176 };
177 let read_only = |reason| refused(io::Error::new(io::ErrorKind::ReadOnlyFilesystem, reason));
178 let id = self.id(path).map_err(refused)?;
179 if !self.provider.document(&id).map_err(refused)?.writable {
180 return Err(read_only(
181 "The app keeping this section lets Snowbound only read it",
182 ));
183 }
184 let seekable = self
185 .opened(&id, "r", |file| Ok(file.stream_position().is_ok()))
186 .map_err(refused)?;
187 // Streamed documents stay read-only until lost publications merge back (plan commit 2).
188 if !seekable {
189 return Err(read_only(
190 "The app keeping this section can't save changes in place yet; they wait on this device",
191 ));
192 }
193 let mut io = Seekable(self.provider.open(&id, "rw").map_err(refused)?);
194 let result = operation(&mut io);
195 let closed = self.provider.close(io.0);
196 result?;
197 closed.map_err(|error| CommitError {
198 state: CommitState::Unknown,
199 error,
200 })
201 }
202
203 /// Renames the document at `path` within its folder.
204 fn rename(&self, path: &str, name: &str) -> io::Result<()> {
205 self.provider.rename(&self.id(path)?, name)?;
206 lock(&self.ids).clear();
207 Ok(())
208 }
209}
210
211fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
212 mutex.lock().unwrap_or_else(PoisonError::into_inner)
213}
214
215fn whole(file: &mut File, limit: usize) -> io::Result<Vec<u8>> {
216 let mut bytes = Vec::new();
217 file.take(limit as u64 + 1).read_to_end(&mut bytes)?;
218 if bytes.len() > limit {
219 return Err(io::ErrorKind::FileTooLarge.into());
220 }
221 Ok(bytes)
222}
223
224/// A seekable descriptor as a commit writes it.
225struct Seekable(File);
226
227impl CommitIo for Seekable {
228 fn read_at(&mut self, offset: u64, bytes: &mut [u8]) -> io::Result<usize> {
229 self.0.seek(SeekFrom::Start(offset))?;
230 self.0.read(bytes)
231 }
232
233 fn write_at(&mut self, offset: u64, bytes: &[u8]) -> io::Result<usize> {
234 self.0.seek(SeekFrom::Start(offset))?;
235 self.0.write(bytes)
236 }
237
238 fn flush(&mut self) -> io::Result<()> {
239 self.0.sync_all()
240 }
241}
242
243/// A section a document provider keeps, as its replica publishes to it.
244pub(crate) struct ProviderRemote {
245 documents: Arc<Documents>,
246 path: String,
247 /// The image the last stamp had to read whole, which the step's read takes.
248 read: Option<Vec<u8>>,
249}
250
251impl ProviderRemote {
252 pub(crate) fn new(documents: &Arc<Documents>, path: &str) -> Self {
253 Self {
254 documents: Arc::clone(documents),
255 path: path.to_owned(),
256 read: None,
257 }
258 }
259}
260
261impl crate::Remote for ProviderRemote {
262 /// A torn or damaged image, as a stream mid-upload reads, is never a base to rebase on.
263 fn read(&mut self) -> io::Result<Vec<u8>> {
264 let image = match self.read.take() {
265 Some(image) => image,
266 None => self.documents.read(&self.path, crate::MAX_FILE_BYTES)?,
267 };
268 if !Store::parse(&image).is_ok_and(|store| store.checksum_mismatches.is_empty()) {
269 return Err(io::Error::new(
270 io::ErrorKind::WouldBlock,
271 "The section was read mid-write",
272 ));
273 }
274 Ok(image)
275 }
276
277 fn stamp(&mut self) -> io::Result<Stamp> {
278 let (stamp, image) = self.documents.stamp(&self.path)?;
279 self.read = image;
280 Ok(stamp)
281 }
282
283 fn publish(&mut self, transaction: &Transaction) -> std::result::Result<(), CommitError> {
284 self.read = None;
285 self.documents
286 .write(&self.path, |io| transaction.commit(io))
287 }
288
289 fn confirm(&mut self, base: &Stamp) -> std::result::Result<(), CommitError> {
290 self.read = None;
291 self.documents
292 .write(&self.path, |io| onestore::confirm(io, base))
293 }
294}
295
296/// A notebook folder a document provider keeps, its folders as last listed kept at `listed`.
297pub(crate) struct Tree {
298 pub(crate) documents: Arc<Documents>,
299 pub(crate) listed: PathBuf,
300}
301
302/// The provider's folders, or as they were last listed while it can't be reached.
303struct Source<'a> {
304 documents: &'a Documents,
305 kept: BTreeMap<String, Vec<discover::Entry>>,
306 listed: BTreeMap<String, Vec<discover::Entry>>,
307 /// Documents sharing a name with one listed, by folder.
308 duplicates: BTreeMap<String, Vec<Document>>,
309}
310
311impl discover::Source for Source<'_> {
312 fn entries(&mut self, path: &str, limit: usize) -> io::Result<Vec<discover::Entry>> {
313 let entries = match self.documents.listing(path) {
314 Ok((entries, duplicates)) => {
315 self.duplicates.insert(path.to_owned(), duplicates);
316 entries
317 }
318 Err(error) if error.kind() == io::ErrorKind::NotConnected => {
319 self.kept.get(path).cloned().ok_or(error)?
320 }
321 Err(error) => return Err(error),
322 };
323 if entries.len() > limit {
324 return Err(io::ErrorKind::FileTooLarge.into());
325 }
326 self.listed.insert(path.to_owned(), entries.clone());
327 Ok(entries)
328 }
329
330 fn read(&mut self, path: &str, limit: usize) -> io::Result<Vec<u8>> {
331 self.documents.read(path, limit)
332 }
333}
334
335fn unsupported() -> Error {
336 io::Error::new(
337 io::ErrorKind::Unsupported,
338 "Snowbound can't change this notebook's folders in another app yet",
339 )
340 .into()
341}
342
343impl Storage for Tree {
344 /// Of several documents with one name, the first by id stands for it and the rest are
345 /// unavailable, named with their ids.
346 fn discover(
347 &self,
348 cache: &mut discover::Cache,
349 limits: discover::Limits,
350 ) -> Result<discover::Folder> {
351 let kept = crate::fs::read(&self.listed)
352 .ok()
353 .and_then(|bytes| serde_json::from_slice(&bytes).ok())
354 .unwrap_or_default();
355 let mut source = Source {
356 documents: &self.documents,
357 kept,
358 listed: BTreeMap::new(),
359 duplicates: BTreeMap::new(),
360 };
361 let mut catalog = cache.discover(&mut source, limits)?;
362 if let Ok(json) = serde_json::to_vec(&source.listed) {
363 let _ = crate::fs::write(&self.listed, json);
364 }
365 for (path, duplicates) in source.duplicates {
366 let Some(folder) = folder(&mut catalog, &path) else {
367 continue;
368 };
369 for duplicate in duplicates {
370 folder.unavailable.push(discover::Unavailable {
371 path: discover::join(&path, &format!("{} ({})", duplicate.name, duplicate.id)),
372 group: duplicate.directory,
373 error: "Another document here has the same name".into(),
374 reason: discover::Reason::Denied,
375 });
376 }
377 }
378 Ok(catalog)
379 }
380
381 fn location(&self) -> String {
382 self.documents.location()
383 }
384
385 fn entries(&self, folder: &str) -> io::Result<Vec<discover::Entry>> {
386 self.documents.entries(folder)
387 }
388
389 fn exists(&self, path: &str) -> bool {
390 let (folder, name) = path.rsplit_once('/').unwrap_or(("", path));
391 self.documents
392 .entries(folder)
393 .is_ok_and(|entries| entries.iter().any(|entry| entry.name == name))
394 }
395
396 fn stamp(&self, path: &str) -> io::Result<Stamp> {
397 Ok(self.documents.stamp(path)?.0)
398 }
399
400 fn read(&self, path: &str) -> Result<Vec<u8>> {
401 Ok(self.documents.read(path, crate::MAX_FILE_BYTES)?)
402 }
403
404 fn read_file(&self, path: &str, limit: usize) -> Result<Vec<u8>> {
405 Ok(self.documents.read(path, limit)?)
406 }
407
408 fn create(&self, _: &str, _: &[u8]) -> Result<()> {
409 Err(unsupported())
410 }
411
412 fn create_directory(&self, _: &str) -> Result<()> {
413 Err(unsupported())
414 }
415
416 fn hide(&self, _: &str) -> Result<()> {
417 Err(unsupported())
418 }
419
420 fn rename(&self, from: &str, to: &str) -> Result<()> {
421 let folder = |path: &str| path.rsplit_once('/').map(|(folder, _)| folder.to_owned());
422 if folder(from) != folder(to) {
423 return Err(unsupported());
424 }
425 let name = to.rsplit('/').next().unwrap_or(to);
426 Ok(self.documents.rename(from, name)?)
427 }
428
429 fn rename_root(&self, _: &str, _: &[String]) -> Result<String> {
430 Err(unsupported())
431 }
432
433 fn replace(&self, _: &str, _: &str) -> Result<()> {
434 Err(unsupported())
435 }
436
437 fn delete(&self, _: &str) -> Result<()> {
438 Err(unsupported())
439 }
440
441 fn place(&self, path: &str, ancestor: [u8; 16], name: &str) -> Result<()> {
442 Ok(self.documents.write(path, |io| {
443 onestore::place(io, ancestor, name).map_err(|error| CommitError {
444 state: CommitState::Unknown,
445 error,
446 })
447 })?)
448 }
449
450 fn commit(&self, path: &str, transaction: &Transaction) -> Result<()> {
451 Ok(self.documents.write(path, |io| transaction.commit(io))?)
452 }
453
454 fn confirm(&self, path: &str, base: &Stamp) -> std::result::Result<(), CommitError> {
455 self.documents.write(path, |io| onestore::confirm(io, base))
456 }
457
458 fn supersede(&self, _: &str, _: &Stamp, _: &str) -> Result<()> {
459 Err(unsupported())
460 }
461}
462
463fn folder<'a>(folder: &'a mut discover::Folder, path: &str) -> Option<&'a mut discover::Folder> {
464 if folder.path == path {
465 return Some(folder);
466 }
467 folder
468 .groups
469 .iter_mut()
470 .find_map(|group| self::folder(group, path))
471}
472
473#[cfg(all(test, unix))]
474mod tests;
crates/notebook/src/provider/tests.rs created+808
......@@ -0,0 +1,808 @@
1//! A fake document provider with real descriptors: seekable files, or pipes as the Android lab
2//! measured stream-only providers (`tools/android/provider-lab`).
3
4use super::*;
5use crate::{
6 EditStatus, Replica,
7 session::{self, Background, Event, Notebook, SyncState, SyncStatus},
8};
9use onestore::{
10 ExGuid,
11 op::{Edit, Op, PageOp},
12 page::{Page, PageObject},
13};
14use std::{
15 fs,
16 ops::Range,
17 os::fd::OwnedFd,
18 path::Path,
19 thread,
20 time::{Duration, Instant},
21};
22
23const TOC: &str = "Open Notebook.onetoc2";
24
25#[derive(Clone, Copy, PartialEq)]
26enum Kind {
27 Seekable,
28 /// Pipes in every mode, `rw` included, with no metadata.
29 Stream,
30}
31
32struct Node {
33 parent: String,
34 name: String,
35 directory: bool,
36}
37
38struct State {
39 kind: Kind,
40 nodes: BTreeMap<String, Node>,
41 next: u32,
42 writable: bool,
43 offline: bool,
44 revoked: bool,
45 /// One device's writes reach the others only at `publish_staged`, as the lab's
46 /// upload-on-close provider showed.
47 delayed: bool,
48 /// The next `rw` open finds its document renamed to `Moved.one` under a new id, and a
49 /// document holding these bytes in its place.
50 swap: Option<Vec<u8>>,
51 modes: Vec<String>,
52}
53
54struct Shared {
55 folder: tempfile::TempDir,
56 state: Mutex<State>,
57}
58
59/// One device's view of the provider.
60#[derive(Clone)]
61struct Fake {
62 store: Arc<Shared>,
63 device: u32,
64}
65
66impl Fake {
67 fn new(kind: Kind) -> Self {
68 let mut nodes = BTreeMap::new();
69 nodes.insert(
70 "root".to_owned(),
71 Node {
72 parent: String::new(),
73 name: String::new(),
74 directory: true,
75 },
76 );
77 let state = State {
78 kind,
79 nodes,
80 next: 0,
81 writable: true,
82 offline: false,
83 revoked: false,
84 delayed: false,
85 swap: None,
86 modes: Vec::new(),
87 };
88 Self {
89 store: Arc::new(Shared {
90 folder: tempfile::tempdir().unwrap(),
91 state: Mutex::new(state),
92 }),
93 device: 0,
94 }
95 }
96
97 fn device(&self, device: u32) -> Self {
98 Self {
99 store: Arc::clone(&self.store),
100 device,
101 }
102 }
103
104 fn documents(&self, root: &str) -> Arc<Documents> {
105 Arc::new(Documents::new(self.clone(), root))
106 }
107
108 fn state(&self) -> MutexGuard<'_, State> {
109 lock(&self.store.state)
110 }
111
112 fn file(&self, id: &str) -> PathBuf {
113 self.store.folder.path().join(id)
114 }
115
116 fn staged(&self, id: &str) -> PathBuf {
117 self.store
118 .folder
119 .path()
120 .join(format!("{id}.{}", self.device))
121 }
122
123 /// The bytes this device reads.
124 fn visible(&self, id: &str) -> PathBuf {
125 let staged = self.staged(id);
126 if staged.exists() {
127 staged
128 } else {
129 self.file(id)
130 }
131 }
132
133 fn add(&self, name: &str, bytes: &[u8]) -> String {
134 let mut state = self.state();
135 state.next += 1;
136 let id = format!("doc-{:04}", state.next);
137 state.nodes.insert(
138 id.clone(),
139 Node {
140 parent: "root".into(),
141 name: name.into(),
142 directory: false,
143 },
144 );
145 fs::write(self.file(&id), bytes).unwrap();
146 id
147 }
148
149 /// The first document named `name` at the root, as stored.
150 fn id(&self, name: &str) -> String {
151 let state = self.state();
152 let (id, _) = state
153 .nodes
154 .iter()
155 .find(|(_, node)| node.parent == "root" && node.name == name)
156 .unwrap();
157 id.clone()
158 }
159
160 fn bytes(&self, name: &str) -> Vec<u8> {
161 fs::read(self.file(&self.id(name))).unwrap()
162 }
163
164 fn set(&self, change: impl FnOnce(&mut State)) {
165 change(&mut self.state());
166 }
167
168 fn modes(&self) -> Vec<String> {
169 self.state().modes.clone()
170 }
171
172 /// Makes every device's writes the document, in device order, as an explicit provider
173 /// publication does.
174 fn publish_staged(&self) {
175 let ids: Vec<String> = self.state().nodes.keys().cloned().collect();
176 for id in ids {
177 for device in 0..4 {
178 let staged = self.device(device).staged(&id);
179 if staged.exists() {
180 fs::rename(staged, self.file(&id)).unwrap();
181 }
182 }
183 }
184 }
185
186 /// Gives the document a new id, as another app renaming it does; the old one is gone.
187 fn reidentify(&self, name: &str) -> String {
188 let old = self.id(name);
189 let mut state = self.state();
190 state.next += 1;
191 let new = format!("doc-{:04}", state.next);
192 let node = state.nodes.remove(&old).unwrap();
193 state.nodes.insert(new.clone(), node);
194 fs::rename(self.file(&old), self.file(&new)).unwrap();
195 new
196 }
197
198 fn reachable(&self) -> io::Result<MutexGuard<'_, State>> {
199 let state = self.state();
200 if state.offline {
201 return Err(io::Error::new(io::ErrorKind::NotConnected, "Offline"));
202 }
203 if state.revoked {
204 return Err(io::Error::new(io::ErrorKind::PermissionDenied, "Revoked"));
205 }
206 Ok(state)
207 }
208
209 fn describe(&self, state: &State, id: &str) -> io::Result<Document> {
210 let node = state.nodes.get(id).ok_or(io::ErrorKind::NotFound)?;
211 let metadata = fs::metadata(self.visible(id)).ok();
212 let known = |value: u64| (state.kind == Kind::Seekable).then_some(value);
213 Ok(Document {
214 id: id.to_owned(),
215 name: node.name.clone(),
216 directory: node.directory,
217 size: metadata.as_ref().and_then(|metadata| known(metadata.len())),
218 modified: metadata
219 .and_then(|metadata| metadata.modified().ok())
220 .and_then(|at| at.duration_since(std::time::UNIX_EPOCH).ok())
221 .and_then(|since| known(since.as_nanos() as u64)),
222 writable: state.writable,
223 })
224 }
225}
226
227impl Provider for Fake {
228 fn document(&self, id: &str) -> io::Result<Document> {
229 let state = self.reachable()?;
230 self.describe(&state, id)
231 }
232
233 fn children(&self, id: &str) -> io::Result<Vec<Document>> {
234 let state = self.reachable()?;
235 if !state
236 .nodes
237 .get(id)
238 .ok_or(io::ErrorKind::NotFound)?
239 .directory
240 {
241 return Err(io::ErrorKind::NotADirectory.into());
242 }
243 let children = state.nodes.iter().filter(|(_, node)| node.parent == id);
244 children.map(|(id, _)| self.describe(&state, id)).collect()
245 }
246
247 fn open(&self, id: &str, mode: &str) -> io::Result<File> {
248 let mut state = self.reachable()?;
249 if !state.nodes.contains_key(id) {
250 return Err(io::ErrorKind::NotFound.into());
251 }
252 state.modes.push(mode.to_owned());
253 if mode != "r" && !state.writable {
254 return Err(io::Error::new(
255 io::ErrorKind::PermissionDenied,
256 "Read-only grant",
257 ));
258 }
259 if mode == "rw"
260 && let Some(twin) = state.swap.take()
261 {
262 let mut node = state.nodes.remove(id).unwrap();
263 let name = std::mem::replace(&mut node.name, "Moved.one".into());
264 let parent = node.parent.clone();
265 let (moved, taken) = (format!("{id}-moved"), format!("{id}-taken"));
266 state.nodes.insert(moved.clone(), node);
267 let directory = false;
268 state.nodes.insert(
269 taken.clone(),
270 Node {
271 parent,
272 name,
273 directory,
274 },
275 );
276 fs::rename(self.file(id), self.file(&moved))?;
277 fs::write(self.file(&taken), twin)?;
278 return Err(io::ErrorKind::NotFound.into());
279 }
280 match (state.kind, mode) {
281 (Kind::Seekable, "r") => File::open(self.visible(id)),
282 (Kind::Seekable, _) => {
283 let mut file = self.file(id);
284 if state.delayed {
285 let staged = self.staged(id);
286 if !staged.exists() {
287 fs::copy(&file, &staged)?;
288 }
289 file = staged;
290 }
291 File::options()
292 .read(true)
293 .write(true)
294 .truncate(mode == "wt")
295 .open(file)
296 }
297 (Kind::Stream, "r") => {
298 let (reader, mut writer) = io::pipe()?;
299 let bytes = fs::read(self.file(id))?;
300 thread::spawn(move || writer.write_all(&bytes));
301 Ok(File::from(OwnedFd::from(reader)))
302 }
303 (Kind::Stream, _) => Ok(File::from(OwnedFd::from(io::pipe()?.1))),
304 }
305 }
306
307 fn close(&self, _: File) -> io::Result<()> {
308 Ok(())
309 }
310
311 fn rename(&self, id: &str, name: &str) -> io::Result<String> {
312 let mut state = self.reachable()?;
313 let mut node = state.nodes.remove(id).ok_or(io::ErrorKind::NotFound)?;
314 node.name = name.to_owned();
315 state.next += 1;
316 let new = format!("doc-{:04}", state.next);
317 state.nodes.insert(new.clone(), node);
318 fs::rename(self.file(id), self.file(&new))?;
319 Ok(new)
320 }
321}
322
323/// A notebook folder: its TOC and two sections it lists, placed under it.
324fn notebook(fake: &Fake) {
325 let mut notes = onestore::create_section("Notes.one", "Original text", "Author").unwrap();
326 let mut ideas = onestore::create_section("Ideas.one", "First idea", "Author").unwrap();
327 let file = |image: &[u8]| Store::parse(image).unwrap().header.file_id;
328 let toc = onestore::create_table_of_contents(
329 TOC,
330 &[("Notes.one", file(&notes)), ("Ideas.one", file(&ideas))],
331 )
332 .unwrap();
333 onestore::place_image(&mut notes, Some((file(&toc), "Notes.one"))).unwrap();
334 onestore::place_image(&mut ideas, Some((file(&toc), "Ideas.one"))).unwrap();
335 fake.add(TOC, &toc);
336 fake.add("Notes.one", &notes);
337 fake.add("Ideas.one", &ideas);
338}
339
340/// Opens a section of `notebook` as a host does: its replica, made from the file only where
341/// there is none.
342fn open(notebook: &Notebook, documents: &Arc<Documents>, path: &str) -> session::Section {
343 let cache = notebook.replica_path(path).unwrap();
344 fs::create_dir_all(cache.parent().unwrap()).unwrap();
345 let replica = Replica::open_or_create(&cache, None, || notebook.read_section(path)).unwrap();
346 session::Section::resume_provider(path.into(), replica, Arc::clone(documents), false, || {})
347 .unwrap()
348}
349
350fn paragraph(page: &Page) -> (ExGuid, String) {
351 page.objects
352 .iter()
353 .find_map(|object| match object {
354 PageObject::Outline(outline) => outline.paragraphs.iter().find_map(|paragraph| {
355 let text = paragraph.text()?;
356 Some((text.id, text.text.text().to_owned()))
357 }),
358 _ => None,
359 })
360 .unwrap()
361}
362
363fn first_page(section: &session::Section) -> (ExGuid, Page) {
364 let space = section.pages().unwrap()[0].0;
365 (space, section.page(space).unwrap())
366}
367
368fn text(section: &session::Section) -> String {
369 paragraph(&first_page(section).1).1
370}
371
372/// Replaces `range` of the first page's text as the editor does, returning the edit's id.
373fn typed(section: &session::Section, range: Range<u32>, with: &str) -> u64 {
374 let (space, page) = first_page(section);
375 let op = PageOp::Text {
376 text: paragraph(&page).0,
377 range,
378 with: with.into(),
379 };
380 let edit = Edit {
381 at: crate::now(),
382 ops: vec![Op::Page { space, op }],
383 };
384 section.replica().apply("Editor", edit).unwrap()
385}
386
387/// The first page's text in `image`, which must cold-open as a section.
388fn stored(image: Vec<u8>) -> String {
389 let arena = onestore::Arena::default();
390 let mut section = onestore::Section::open(&arena, image).unwrap();
391 let space = section.pages().unwrap()[0].0;
392 paragraph(&section.page(space).unwrap()).1
393}
394
395fn until(what: &str, mut done: impl FnMut() -> bool) {
396 let deadline = Instant::now() + Duration::from_secs(20);
397 while !done() {
398 assert!(Instant::now() < deadline, "{what}");
399 thread::sleep(Duration::from_millis(20));
400 }
401}
402
403fn published(section: &session::Section, id: u64) {
404 until("the edit was published", || {
405 section.wake();
406 matches!(
407 section.status(id).unwrap(),
408 Some(EditStatus::Published { .. })
409 )
410 });
411}
412
413fn status(section: &session::Section, accept: impl Fn(&SyncStatus) -> bool) -> SyncStatus {
414 let mut last = None;
415 until("the status came", || {
416 section.wake();
417 let status = section.sync_status().unwrap();
418 let accepted = accept(&status);
419 last = Some(status);
420 accepted
421 });
422 last.unwrap()
423}
424
425#[test]
426fn seekable_documents_take_commits_in_place_and_rebase_on_another_devices() {
427 let fake = Fake::new(Kind::Seekable);
428 notebook(&fake);
429 let cache = tempfile::tempdir().unwrap();
430 let documents = fake.documents("root");
431 let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
432 assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]);
433 let section = open(&notebook, &documents, "Notes.one");
434 let before = fake.bytes("Notes.one").len();
435 let id = typed(&section, 0..0, "Hello ");
436 published(&section, id);
437 let after = fake.bytes("Notes.one");
438 assert!(after.len() > before, "the revision was appended");
439 assert_eq!(stored(after), "Hello Original text");
440 assert!(fake.modes().iter().all(|mode| mode == "r" || mode == "rw"));
441
442 let elsewhere = fake.device(1).documents("root");
443 let other_cache = tempfile::tempdir().unwrap();
444 let other = Notebook::open_provider(Arc::clone(&elsewhere), other_cache.path()).unwrap();
445 let other = open(&other, &elsewhere, "Notes.one");
446 let id = typed(&other, 0..0, "World ");
447 published(&other, id);
448 until("the change reached the first device", || {
449 section.wake();
450 text(&section) == "World Hello Original text"
451 });
452 assert!(
453 section
454 .events()
455 .iter()
456 .any(|event| matches!(event, Event::Changed(_)))
457 );
458 assert_eq!(stored(fake.bytes("Notes.one")), "World Hello Original text");
459 section.close().unwrap();
460 other.close().unwrap();
461}
462
463#[test]
464fn stream_only_documents_are_read_and_never_opened_to_write() {
465 let fake = Fake::new(Kind::Stream);
466 notebook(&fake);
467 let cache = tempfile::tempdir().unwrap();
468 let documents = fake.documents("root");
469 let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
470 let section = open(&notebook, &documents, "Notes.one");
471 let id = typed(&section, 13..13, " mine");
472 let waiting = status(&section, |status| status.error.is_some());
473 assert_eq!(waiting.state(), SyncState::ReadOnly);
474 assert_eq!(waiting.queued, 1);
475 assert!(fake.modes().iter().all(|mode| mode == "r"));
476
477 // Another device, which can write in place, changes the document.
478 let elsewhere = Fake::new(Kind::Seekable);
479 for name in [TOC, "Notes.one", "Ideas.one"] {
480 elsewhere.add(name, &fake.bytes(name));
481 }
482 let other_documents = elsewhere.documents("root");
483 let other_cache = tempfile::tempdir().unwrap();
484 let other = Notebook::open_provider(Arc::clone(&other_documents), other_cache.path()).unwrap();
485 let other = open(&other, &other_documents, "Notes.one");
486 let theirs = typed(&other, 0..0, "Theirs ");
487 published(&other, theirs);
488 other.close().unwrap();
489 fs::write(
490 fake.file(&fake.id("Notes.one")),
491 elsewhere.bytes("Notes.one"),
492 )
493 .unwrap();
494 until("the change was read", || {
495 section.wake();
496 text(&section) == "Theirs Original text mine"
497 });
498 assert_eq!(section.pending().unwrap()[0].id, id);
499 assert_eq!(stored(fake.bytes("Notes.one")), "Theirs Original text");
500 assert!(fake.modes().iter().all(|mode| mode == "r"));
501 section.close().unwrap();
502}
503
504#[test]
505fn documents_without_a_write_grant_are_only_read() {
506 let fake = Fake::new(Kind::Seekable);
507 notebook(&fake);
508 fake.set(|state| state.writable = false);
509 let cache = tempfile::tempdir().unwrap();
510 let documents = fake.documents("root");
511 let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
512 let section = open(&notebook, &documents, "Notes.one");
513 let before = fake.bytes("Notes.one");
514 typed(&section, 0..0, "Mine ");
515 let status = status(&section, |status| status.error.is_some());
516 assert_eq!(status.state(), SyncState::ReadOnly);
517 assert_eq!(
518 status.error.unwrap().kind(),
519 io::ErrorKind::ReadOnlyFilesystem
520 );
521 assert_eq!(status.queued, 1);
522 assert_eq!(fake.bytes("Notes.one"), before);
523 assert!(fake.modes().iter().all(|mode| mode == "r"));
524 section.close().unwrap();
525}
526
527#[test]
528fn a_write_another_device_sees_late_rebases_it_once_visible() {
529 let fake = Fake::new(Kind::Seekable);
530 notebook(&fake);
531 fake.set(|state| state.delayed = true);
532 let (writer, reader) = (fake.documents("root"), fake.device(1).documents("root"));
533 let caches = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
534 let writing = Notebook::open_provider(Arc::clone(&writer), caches.0.path()).unwrap();
535 let reading = Notebook::open_provider(Arc::clone(&reader), caches.1.path()).unwrap();
536 let (writing, reading) = (
537 open(&writing, &writer, "Notes.one"),
538 open(&reading, &reader, "Notes.one"),
539 );
540 let id = typed(&writing, 0..0, "Early ");
541 published(&writing, id);
542 // The writer's close succeeded; the other device still reads the old bytes.
543 let old = reader.read("Notes.one", crate::MAX_FILE_BYTES).unwrap();
544 assert_eq!(stored(old), "Original text");
545 reading.wake();
546 thread::sleep(Duration::from_millis(200));
547 assert_eq!(text(&reading), "Original text");
548 reading.set_offline(true);
549 let id = typed(&reading, 13..13, " later");
550 fake.publish_staged();
551 reading.set_offline(false);
552 published(&reading, id);
553 assert_eq!(text(&reading), "Early Original text later");
554 fake.publish_staged();
555 assert_eq!(stored(fake.bytes("Notes.one")), "Early Original text later");
556 writing.close().unwrap();
557 reading.close().unwrap();
558}
559
560#[test]
561fn a_cold_start_while_offline_opens_the_last_listing_and_queued_edits() {
562 let fake = Fake::new(Kind::Seekable);
563 notebook(&fake);
564 let cache = tempfile::tempdir().unwrap();
565 let documents = fake.documents("root");
566 let mut notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
567 let background = Background::provider(Arc::clone(&documents), || {}).unwrap();
568 background.watch(notebook.replicas());
569 until("every section has its offline copy", || {
570 ["Notes.one", "Ideas.one"]
571 .iter()
572 .all(|path| notebook.replica_path(path).unwrap().exists())
573 });
574 background.stop();
575 let section = open(&notebook, &documents, "Ideas.one");
576 fake.set(|state| state.offline = true);
577 let id = typed(&section, 0..0, "Offline ");
578 status(&section, |status| {
579 status.state() == SyncState::NotConnected && status.queued == 1
580 });
581 section.close().unwrap();
582 drop(notebook);
583
584 // A new process: the provider fails throughout.
585 let documents = fake.documents("root");
586 let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
587 assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]);
588 let notes = open(&notebook, &documents, "Notes.one");
589 assert_eq!(text(&notes), "Original text");
590 let section = open(&notebook, &documents, "Ideas.one");
591 assert_eq!(text(&section), "Offline First idea");
592 assert_eq!(section.pending().unwrap()[0].id, id);
593 status(&section, |status| status.state() == SyncState::NotConnected);
594 fake.set(|state| state.offline = false);
595 published(&section, id);
596 assert_eq!(stored(fake.bytes("Ideas.one")), "Offline First idea");
597 notes.close().unwrap();
598 section.close().unwrap();
599}
600
601#[test]
602fn a_revoked_grant_is_not_offline_and_keeps_the_replicas() {
603 let fake = Fake::new(Kind::Seekable);
604 notebook(&fake);
605 let cache = tempfile::tempdir().unwrap();
606 let documents = fake.documents("root");
607 let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
608 let replica = notebook.replica_path("Notes.one").unwrap();
609 let section = open(&notebook, &documents, "Notes.one");
610 fake.set(|state| state.revoked = true);
611 let id = typed(&section, 0..0, "Kept ");
612 let status = status(&section, |status| status.error.is_some());
613 assert_eq!(
614 status.error.unwrap().kind(),
615 io::ErrorKind::PermissionDenied
616 );
617 section.close().unwrap();
618 drop(notebook);
619 let refused = Notebook::open_provider(fake.documents("root"), cache.path()).err();
620 assert!(
621 matches!(
622 &refused,
623 Some(Error::Discovery(discover::Error::Io { error, .. }))
624 if error.kind() == io::ErrorKind::PermissionDenied
625 ),
626 "{refused:?}"
627 );
628 assert_eq!(
629 Replica::open(&replica).unwrap().pending().unwrap()[0].id,
630 id
631 );
632
633 fake.set(|state| state.revoked = false);
634 let documents = fake.documents("root");
635 let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
636 let section = open(&notebook, &documents, "Notes.one");
637 published(&section, id);
638 assert_eq!(stored(fake.bytes("Notes.one")), "Kept Original text");
639 section.close().unwrap();
640}
641
642#[test]
643fn documents_renamed_to_new_ids_are_listed_again() {
644 let fake = Fake::new(Kind::Seekable);
645 notebook(&fake);
646 let cache = tempfile::tempdir().unwrap();
647 let documents = fake.documents("root");
648 let mut notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
649 let section = open(&notebook, &documents, "Notes.one");
650 let id = typed(&section, 0..0, "One ");
651 published(&section, id);
652 fake.reidentify("Notes.one");
653 let id = typed(&section, 0..0, "Two ");
654 published(&section, id);
655 assert_eq!(stored(fake.bytes("Notes.one")), "Two One Original text");
656 section.close().unwrap();
657
658 let old = fake.id("Notes.one");
659 assert_eq!(
660 notebook.rename("Notes.one", "Journal").unwrap(),
661 "Journal.one"
662 );
663 assert!(matches!(
664 fake.document(&old).map_err(|error| error.kind()),
665 Err(io::ErrorKind::NotFound)
666 ));
667 assert_eq!(notebook.catalog().order, ["Journal.one", "Ideas.one"]);
668 let identity = notebook.catalog().sections[0].file_id;
669 assert!(session::lists(&fake.bytes(TOC), identity).unwrap());
670 let section = open(&notebook, &documents, "Journal.one");
671 let id = typed(&section, 0..0, "Three ");
672 published(&section, id);
673 assert_eq!(
674 stored(fake.bytes("Journal.one")),
675 "Three Two One Original text"
676 );
677 section.close().unwrap();
678}
679
680#[test]
681fn of_documents_sharing_a_name_one_lists_and_the_rest_are_unavailable() {
682 let fake = Fake::new(Kind::Seekable);
683 notebook(&fake);
684 let twin = fake.add(
685 "Notes.one",
686 &onestore::create_section("Notes.one", "A twin", "Author").unwrap(),
687 );
688 let cache = tempfile::tempdir().unwrap();
689 let documents = fake.documents("root");
690 let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
691 assert_eq!(notebook.catalog().order, ["Notes.one", "Ideas.one"]);
692 let unavailable = &notebook.catalog().unavailable;
693 assert_eq!(unavailable.len(), 1);
694 assert_eq!(unavailable[0].path, format!("Notes.one ({twin})"));
695 let section = open(&notebook, &documents, "Notes.one");
696 assert_eq!(text(&section), "Original text");
697 section.close().unwrap();
698}
699
700#[test]
701fn a_document_taking_the_name_between_check_and_write_is_not_written() {
702 let fake = Fake::new(Kind::Seekable);
703 notebook(&fake);
704 let cache = tempfile::tempdir().unwrap();
705 let documents = fake.documents("root");
706 let notebook = Notebook::open_provider(Arc::clone(&documents), cache.path()).unwrap();
707 let section = open(&notebook, &documents, "Notes.one");
708 let id = typed(&section, 0..0, "Checked ");
709 published(&section, id);
710 let original = fake.bytes("Notes.one");
711 let twin = onestore::create_section("Notes.one", "A twin", "Author").unwrap();
712 fake.set(|state| state.swap = Some(twin.clone()));
713 typed(&section, 0..0, "Refused ");
714 until("the write was refused", || {
715 section.wake();
716 fake.state().swap.is_none() && section.sync_status().unwrap().error.is_some()
717 });
718 for _ in 0..3 {
719 section.wake();
720 thread::sleep(Duration::from_millis(100));
721 }
722 assert_eq!(fake.bytes("Notes.one"), twin);
723 assert_eq!(fake.bytes("Moved.one"), original);
724 // Another section under the name stops the worker, as it does for a file.
725 assert!(section.close().is_err());
726}
727
728#[test]
729fn place_and_confirm_write_a_seekable_document_in_place() {
730 let fake = Fake::new(Kind::Seekable);
731 notebook(&fake);
732 let listed = tempfile::tempdir().unwrap();
733 let tree = Tree {
734 documents: fake.documents("root"),
735 listed: listed.path().join("entries.json"),
736 };
737 let before = fake.bytes("Ideas.one");
738 let stamp = Stamp::of(&before).unwrap();
739 tree.confirm("Ideas.one", &stamp).unwrap();
740 let confirmed = fake.bytes("Ideas.one");
741 assert_eq!(confirmed.len(), before.len());
742 assert_ne!(confirmed[212..252], before[212..252]);
743 assert_eq!(stored(confirmed.clone()), "First idea");
744 let stale = tree.confirm("Ideas.one", &stamp).unwrap_err();
745 assert_eq!(stale.state, CommitState::NotCommitted);
746 tree.place("Ideas.one", [7; 16], "Renamed.one").unwrap();
747 let mut placed = confirmed;
748 onestore::place_image(&mut placed, Some(([7; 16], "Renamed.one"))).unwrap();
749 assert_eq!(fake.bytes("Ideas.one"), placed);
750 assert!(fake.modes().iter().all(|mode| mode == "r" || mode == "rw"));
751}
752
753#[test]
754fn conflicting_edits_leave_a_conflict_page_in_the_document() {
755 let fake = Fake::new(Kind::Seekable);
756 notebook(&fake);
757 let (mine, theirs) = (fake.documents("root"), fake.device(1).documents("root"));
758 let caches = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
759 let local = Notebook::open_provider(Arc::clone(&mine), caches.0.path()).unwrap();
760 let remote = Notebook::open_provider(Arc::clone(&theirs), caches.1.path()).unwrap();
761 let (local, remote) = (
762 open(&local, &mine, "Notes.one"),
763 open(&remote, &theirs, "Notes.one"),
764 );
765 local.set_offline(true);
766 typed(&local, 0..8, "Local");
767 let id = typed(&remote, 0..8, "Native");
768 published(&remote, id);
769 local.set_offline(false);
770 until("the conflict page was published", || {
771 local.wake();
772 !local.conflicts().unwrap().is_empty() && local.pending().unwrap().is_empty()
773 });
774 assert_eq!(text(&local), "Native text");
775 let conflicts = local.conflicts().unwrap();
776 let kept = local.page(conflicts[0].1[0].space).unwrap();
777 assert_eq!(paragraph(&kept).1, "Local text");
778 let arena = onestore::Arena::default();
779 let mut stored = onestore::Section::open(&arena, fake.bytes("Notes.one")).unwrap();
780 assert_eq!(stored.conflicts().unwrap().len(), 1);
781 local.close().unwrap();
782 remote.close().unwrap();
783}
784
785#[test]
786fn a_lone_document_opens_offline_from_its_replica() {
787 let fake = Fake::new(Kind::Seekable);
788 let image = onestore::create_section("Lone.one", "Alone", "Author").unwrap();
789 let lone = fake.add("Lone.one", &image);
790 let cache = tempfile::tempdir().unwrap();
791 let open = |fake: &Fake| {
792 session::Section::open_provider(fake.documents(&lone), cache.path(), false, || {}).unwrap()
793 };
794 let section = open(&fake);
795 let id = typed(&section, 0..0, "Not ");
796 published(&section, id);
797 section.close().unwrap();
798 fake.set(|state| state.offline = true);
799 let section = open(&fake);
800 assert_eq!(section.file(), Path::new(&lone));
801 assert_eq!(text(&section), "Not Alone");
802 let id = typed(&section, 9..9, " now");
803 status(&section, |status| status.state() == SyncState::NotConnected);
804 fake.set(|state| state.offline = false);
805 published(&section, id);
806 assert_eq!(stored(fake.bytes("Lone.one")), "Not Alone now");
807 section.close().unwrap();
808}
crates/notebook/src/session.rs+126-13
......@@ -514,6 +514,21 @@ impl Notebook {
514514 )
515515 }
516516
517 /// Opens the notebook whose folder is the root of `documents`. Sections open through
518 /// `Section::resume_provider` with the catalog's paths; while the provider can't be
519 /// reached, the notebook opens as its folders were last listed.
520 pub fn open_provider(
521 documents: Arc<crate::provider::Documents>,
522 cache: impl AsRef<Path>,
523 ) -> Result<Self> {
524 let listed = listing(cache.as_ref(), &documents.location()).with_extension("entries.json");
525 Self::with(
526 Box::new(crate::provider::Tree { documents, listed }),
527 None,
528 cache,
529 )
530 }
531
517532 /// The storage the notebook's files are in, for a Live Share host to serve.
518533 pub fn into_storage(self) -> Box<dyn Storage> {
519534 self.storage
......@@ -1844,14 +1859,21 @@ impl Notebook {
18441859 if fs::metadata(&cache).is_ok() {
18451860 let (replica, remote, mut connect) =
18461861 (Replica::open(&cache)?, file.clone(), connect);
1847 return Section::start(file, replica, move || connect(&remote), offline, notify);
1862 return Section::start(
1863 file,
1864 replica,
1865 move || connect(&remote),
1866 CHECK,
1867 offline,
1868 notify,
1869 );
18481870 }
18491871 }
18501872 let replica = |identity: &[u8; 16], _: &[u8]| {
18511873 fs::create_dir_all(&replicas)?;
18521874 Ok(replica_file(&replicas, identity))
18531875 };
1854 Section::open_in(file, key, replica, connect, offline, notify)
1876 Section::open_in(file, key, replica, connect, CHECK, offline, notify)
18551877 }
18561878}
18571879
......@@ -2110,6 +2132,22 @@ fn remove_replica(replica: &Path) -> io::Result<()> {
21102132 Ok(())
21112133}
21122134
2135/// How often an open section's file is checked where nothing watches it.
2136const CHECK: Duration = Duration::from_secs(2);
2137
2138/// Names a lone section's replica in the folder of `location` by its identity, taking one a
2139/// moved file left behind (`location::claim`).
2140fn claimed<'a>(
2141 cache: &'a Path,
2142 location: &'a str,
2143) -> impl FnOnce(&[u8; 16], &[u8]) -> Result<PathBuf> + 'a {
2144 move |identity, source| {
2145 let folder =
2146 crate::location::claim(cache, location, &[*identity], |_| Stamp::of(source).ok())?;
2147 Ok(replica_file(&folder, identity))
2148 }
2149}
2150
21132151/// The replica in `cache` of the section `identity` names.
21142152pub(crate) fn replica_file(cache: &Path, identity: &[u8; 16]) -> PathBuf {
21152153 let name: String = identity.iter().map(|byte| format!("{byte:02x}")).collect();
......@@ -2264,13 +2302,8 @@ impl Section {
22642302 ) -> Result<Self> {
22652303 let file = fs::canonicalize(file)?;
22662304 let location = file.to_string_lossy().into_owned();
2267 let replicas = |identity: &[u8; 16], source: &[u8]| {
2268 let folder = crate::location::claim(cache.as_ref(), &location, &[*identity], |_| {
2269 Stamp::of(source).ok()
2270 })?;
2271 Ok(replica_file(&folder, identity))
2272 };
2273 Self::open_in(file, None, replicas, connect, offline, notify)
2305 let replicas = claimed(cache.as_ref(), &location);
2306 Self::open_in(file, None, replicas, connect, CHECK, offline, notify)
22742307 }
22752308
22762309 /// Opens the canonical section `file`, a protected one under `key`, through the replica
......@@ -2280,6 +2313,7 @@ impl Section {
22802313 key: Option<&protected::Key>,
22812314 replica: impl FnOnce(&[u8; 16], &[u8]) -> Result<PathBuf>,
22822315 mut connect: impl FnMut(&Path) -> io::Result<R> + Send + 'static,
2316 interval: Duration,
22832317 offline: bool,
22842318 notify: impl Fn() + Send + 'static,
22852319 ) -> Result<Self> {
......@@ -2289,7 +2323,14 @@ impl Section {
22892323 let cache = replica(&identity.guid, &source)?;
22902324 let replica = Replica::open_or_create(&cache, key, || Ok(source))?;
22912325 let remote = file.clone();
2292 Self::start(file, replica, move || connect(&remote), offline, notify)
2326 Self::start(
2327 file,
2328 replica,
2329 move || connect(&remote),
2330 interval,
2331 offline,
2332 notify,
2333 )
22932334 }
22942335
22952336 /// Resumes an owned local replica without reading the publication target.
......@@ -2306,6 +2347,7 @@ impl Section {
23062347 file,
23072348 replica,
23082349 move || Ok(FileRemote(remote.clone())),
2350 CHECK,
23092351 offline,
23102352 notify,
23112353 )
......@@ -2326,6 +2368,7 @@ impl Section {
23262368 PathBuf::from(&path),
23272369 replica,
23282370 move || Ok(crate::SmbRemote::new(connect()?, path.clone(), limit)),
2371 CHECK,
23292372 offline,
23302373 notify,
23312374 )
......@@ -2344,15 +2387,84 @@ impl Section {
23442387 PathBuf::from(&path),
23452388 replica,
23462389 move || Ok(crate::live::share::HostedRemote::new(&guest, &path)),
2390 CHECK,
23472391 offline,
23482392 notify,
23492393 )
23502394 }
23512395
2396 /// Resumes a replica against the section at catalog `path` of a notebook a document
2397 /// provider keeps (`Notebook::open_provider`). Its file is checked every
2398 /// `Background::UNWATCHED`, and when `wake` passes on what the host hears of changes.
2399 pub fn resume_provider(
2400 path: String,
2401 replica: Replica,
2402 documents: Arc<crate::provider::Documents>,
2403 offline: bool,
2404 notify: impl Fn() + Send + 'static,
2405 ) -> Result<Self> {
2406 Self::start(
2407 PathBuf::from(&path),
2408 replica,
2409 move || Ok(crate::provider::ProviderRemote::new(&documents, &path)),
2410 Background::UNWATCHED,
2411 offline,
2412 notify,
2413 )
2414 }
2415
2416 /// Opens the lone section that is the root of `documents` as `open` opens a file, checked
2417 /// as `resume_provider`'s is; while the provider can't be reached, from the only replica
2418 /// its location has.
2419 pub fn open_provider(
2420 documents: Arc<crate::provider::Documents>,
2421 cache: impl AsRef<Path>,
2422 offline: bool,
2423 notify: impl Fn() + Send + Sync + 'static,
2424 ) -> Result<Self> {
2425 let (cache, location) = (cache.as_ref(), documents.location());
2426 let file = PathBuf::from(&documents.root);
2427 let remote = move || Ok(crate::provider::ProviderRemote::new(&documents, ""));
2428 let notify = Arc::new(notify);
2429 let (connect, woken) = (remote.clone(), Arc::clone(&notify));
2430 let replicas = claimed(cache, &location);
2431 let opened = Self::open_in(
2432 file.clone(),
2433 None,
2434 replicas,
2435 move |_| connect(),
2436 Background::UNWATCHED,
2437 offline,
2438 move || woken(),
2439 );
2440 match opened {
2441 Err(Error::Io(error)) if error.kind() == io::ErrorKind::NotConnected => {
2442 let mut replicas = fs::read_dir(crate::location::folder(cache, &location))
2443 .into_iter()
2444 .flatten()
2445 .filter_map(|entry| Some(entry.ok()?.path()))
2446 .filter(|path| path.extension() == Some("sqlite".as_ref()));
2447 match (replicas.next(), replicas.next()) {
2448 (Some(only), None) => Self::start(
2449 file,
2450 Replica::open(only)?,
2451 remote,
2452 Background::UNWATCHED,
2453 offline,
2454 move || notify(),
2455 ),
2456 _ => Err(Error::Io(error)),
2457 }
2458 }
2459 opened => opened,
2460 }
2461 }
2462
23522463 fn start<R: Remote + 'static>(
23532464 file: PathBuf,
23542465 replica: Replica,
23552466 connect: impl FnMut() -> io::Result<R> + Send + 'static,
2467 interval: Duration,
23562468 offline: bool,
23572469 notify: impl Fn() + Send + 'static,
23582470 ) -> Result<Self> {
......@@ -2374,20 +2486,21 @@ impl Section {
23742486 notify,
23752487 observed,
23762488 };
2377 section.worker = Some(section.start_worker(connect, offline, None)?);
2489 section.worker = Some(section.start_worker(connect, interval, offline, None)?);
23782490 Ok(section)
23792491 }
23802492
23812493 fn start_worker<R: Remote + 'static>(
23822494 &self,
23832495 connect: impl FnMut() -> io::Result<R> + Send + 'static,
2496 interval: Duration,
23842497 offline: bool,
23852498 replacing: Option<&SyncWorker>,
23862499 ) -> Result<SyncWorker> {
23872500 let (sender, notify) = (self.sender.clone(), Arc::clone(&self.notify));
23882501 let observed = Arc::clone(&self.observed);
23892502 Ok(self.replica.start_sync_replacing(
2390 Duration::from_secs(2),
2503 interval,
23912504 offline,
23922505 connect,
23932506 move |result| {
......@@ -2455,7 +2568,7 @@ impl Section {
24552568 }
24562569 self.observed = Arc::new(Mutex::new(None));
24572570 self.worker =
2458 Some(self.start_worker(move || connect(&target), false, previous.as_ref())?);
2571 Some(self.start_worker(move || connect(&target), CHECK, false, previous.as_ref())?);
24592572 self.file = file;
24602573 Ok(())
24612574 }
crates/notebook/src/session/tests.rs+1
......@@ -568,6 +568,7 @@ fn initially_offline_workers_keep_the_original_queue_until_explicitly_woken() {
568568 calls: Arc::clone(&counter),
569569 })
570570 },
571 CHECK,
571572 true,
572573 move || {
573574 let _ = notify.send(());