diff --git a/crates/notebook/src/background.rs b/crates/notebook/src/background.rs index c87b9d23cf803520dd166395df339ba349bf74a4..f0e57294928e541930e8aabcfdf7b62619ff733d 100644 --- a/crates/notebook/src/background.rs +++ b/crates/notebook/src/background.rs @@ -6,7 +6,7 @@ use crate::fs; use crate::{ EditStatus, Error, Remote, Replica, Result, discover::{Entry, Listed}, - session::{Section, SyncStatus, reached}, + session::{Observed, Section, SyncStatus, reached, sync_status}, worker::Signal, }; use onestore::Stamp; @@ -112,8 +112,14 @@ struct Watch { image: Option>, /// When the section is next checked. due: Instant, - /// The worker of the session that holds the section. - held: Option>, + held: Option, +} + +#[derive(Clone)] +struct Held { + worker: Weak, + replica: Weak, + observed: Weak>, } impl Background { @@ -429,7 +435,11 @@ impl Background { let reported = watched.reported; if let Some(watch) = watched.sections.iter_mut().find(|watch| watch.path == path) { worker.watched.store(reported, Ordering::Release); - watch.held = Some(Arc::downgrade(&worker)); + watch.held = Some(Held { + worker: Arc::downgrade(&worker), + replica: Arc::downgrade(replica), + observed: Arc::downgrade(§ion.observed), + }); } } let (shared, path) = (Arc::downgrade(&self.0), path.to_owned()); @@ -449,10 +459,9 @@ impl Background { self.0.touched(paths); } - /// Each watched section's status as its last check left it, in watch order. A section a - /// session holds keeps the status it had before. + /// Each watched section's status, in watch order; held sections report their live state. pub fn status(&self) -> Vec<(String, SyncStatus)> { - self.0.watched.lock().map_or_else( + let sections: Vec<_> = self.0.watched.lock().map_or_else( |_| Vec::new(), |watched| { watched @@ -470,11 +479,34 @@ impl Background { .map(|error| io::Error::new(error.kind(), error.to_string())), queued: status.queued, }, + watch.held.clone(), ) }) .collect() }, - ) + ); + sections + .into_iter() + .map(|(path, fallback, held)| { + let live = held.and_then(|held| { + let worker = held.worker.upgrade()?; + let replica = held.replica.upgrade()?; + let observed = held.observed.upgrade()?; + let synced = + Some(worker.synced.load(Ordering::Acquire)).filter(|time| *time != 0); + Some(sync_status(&replica, synced, &observed)) + }); + let status = match live { + Some(Ok(status)) => status, + Some(Err(error)) => SyncStatus { + error: Some(reached(&error)), + ..fallback + }, + None => fallback, + }; + (path, status) + }) + .collect() } /// Sections whose file changed since the last call, by catalog path. @@ -678,7 +710,7 @@ impl Watched { for signal in self .sections .iter() - .filter_map(|watch| watch.held.as_ref()?.upgrade()) + .filter_map(|watch| watch.held.as_ref()?.worker.upgrade()) { signal.watched.store(reported, Ordering::Release); signal.wake(); @@ -807,7 +839,7 @@ impl Watched { if !bound { return Next::Connect(self.connection); } - if let Some(worker) = watch.held.as_ref().and_then(Weak::upgrade) { + if let Some(worker) = watch.held.as_ref().and_then(|held| held.worker.upgrade()) { worker.wake(); watch.due = now + interval; continue; @@ -1091,7 +1123,11 @@ mod tests { let mut watched = Watched::default(); watched.watch(sections(1, true), now); let (held, woken) = Signal::new(); - watched.sections[0].held = Some(Arc::downgrade(&held)); + watched.sections[0].held = Some(Held { + worker: Arc::downgrade(&held), + replica: Weak::new(), + observed: Weak::new(), + }); assert!(matches!(watched.next(now, false), Next::Connect(0))); watched.watching(true); assert!(held.watched.load(Ordering::Acquire)); @@ -1430,6 +1466,116 @@ mod tests { drop(rewatch); } + #[test] + fn held_status_tracks_its_worker_without_retaining_the_replica() { + use crate::session::SyncState; + use onestore::{ + PageCreation, + op::{Edit, Op, SectionOp}, + }; + + let directory = tempfile::tempdir().unwrap(); + let file = directory.path().join("Held.one"); + let source = onestore::create_section("Held.one", "", "Author").unwrap(); + fs::write(&file, &source).unwrap(); + let (changed, changes) = mpsc::channel(); + let section = Section::open(&file, directory.path().join("cache"), move || { + let _ = changed.send(()); + }) + .unwrap(); + changes.recv_timeout(Duration::from_secs(3)).unwrap(); + section.set_offline(true); + + let (stamps, _) = mpsc::channel(); + let files = Files { + images: Arc::default(), + stamps, + }; + files.write("Held.one", source); + let remote = files.clone(); + let background = Background::start( + false, + move |_| { + let (files, listed) = (remote.clone(), remote.clone()); + let bind = move |path: &str| File(files.clone(), path.to_owned()); + let list = move |folder: String| std::future::ready(Ok(listed.list(&folder))); + Ok(((bind, list), true)) + }, + || {}, + ) + .unwrap(); + background.set_offline(true); + background.watch(vec![files.known("Held.one")]); + background.hold("Held.one", §ion); + section + .apply( + "Author", + Edit { + at: crate::now(), + ops: vec![Op::Section(SectionOp::Create( + PageCreation::new(None, Some("Queued"), "Author").unwrap(), + ))], + }, + ) + .unwrap(); + let pending = section.pending().unwrap(); + assert_eq!(pending.len(), 1); + let status = background.status().remove(0).1; + assert_eq!(status.queued, 1); + assert!(status.synced.is_some()); + assert_eq!(status.state(), SyncState::Syncing); + { + let watched = background.0.watched.lock().unwrap(); + assert!(watched.sections[0].status.synced.is_none()); + assert_eq!(watched.sections[0].status.queued, 0); + } + + let wait = |expected| { + let deadline = Instant::now() + Duration::from_secs(3); + loop { + let status = background.status().remove(0).1; + if status.state() == expected { + return status; + } + changes + .recv_timeout(deadline.saturating_duration_since(Instant::now())) + .unwrap(); + } + }; + section.set_offline(false); + assert_eq!(wait(SyncState::UpToDate).queued, 0); + assert!(matches!( + section.status(pending[0].id).unwrap(), + Some(EditStatus::Published { .. }) + )); + + let published = fs::read_file(&file).unwrap(); + fs::remove_file(&file).unwrap(); + section.wake(); + assert_eq!( + wait(SyncState::NotConnected).error.unwrap().kind(), + io::ErrorKind::NotFound + ); + fs::write(&file, published).unwrap(); + section.wake(); + assert!(wait(SyncState::UpToDate).error.is_none()); + assert!( + background.0.watched.lock().unwrap().sections[0] + .status + .synced + .is_none() + ); + + let replica = Arc::downgrade(section.replica()); + section.close().unwrap(); + assert!(replica.upgrade().is_none()); + assert!( + background.0.watched.lock().unwrap().sections[0] + .held + .is_none() + ); + } + #[test] fn a_held_section_is_left_to_its_worker_which_the_watch_wakes() { let (sender, stamps) = mpsc::channel(); @@ -1462,7 +1608,11 @@ mod tests { // As `hold` leaves it to a session's worker. let (signal, woken) = Signal::new(); if let Ok(mut watched) = background.0.watched.lock() { - watched.sections[0].held = Some(Arc::downgrade(&signal)); + watched.sections[0].held = Some(Held { + worker: Arc::downgrade(&signal), + replica: Weak::new(), + observed: Weak::new(), + }); signal.watched.store(watched.reported, Ordering::Release); } assert!( diff --git a/crates/notebook/src/session.rs b/crates/notebook/src/session.rs index ad7371f978f0a60d002f66cf3f1622d1b489f1ee..54032a36d89a7bc614e209aff0c32ef03f435595 100644 --- a/crates/notebook/src/session.rs +++ b/crates/notebook/src/session.rs @@ -2084,7 +2084,30 @@ impl SyncStatus { } /// The last attempt's outcome; `None` before the first. -type Observed = Option>; +pub(crate) type Observed = Option>; + +pub(crate) fn sync_status( + replica: &Replica, + synced: Option, + observed: &Mutex, +) -> Result { + let queued: u64 = crate::lock(&replica.section.connection)?.query_row( + "SELECT count(*) FROM edits", + [], + |row| row.get(0), + )?; + let observed = observed + .lock() + .map_err(|_| io::Error::other("Synchronization observer panicked"))?; + Ok(SyncStatus { + synced, + error: observed + .as_ref() + .and_then(Option::as_ref) + .map(|error| io::Error::new(error.kind(), error.to_string())), + queued, + }) +} /// A section file with its replica and background publication. `Send + Sync`: edits /// arrive from any thread without waiting, and `notify` wakes the host when events wait. @@ -2095,7 +2118,7 @@ pub struct Section { events: Mutex>, sender: Sender, notify: Notify, - observed: Arc>, + pub(crate) observed: Arc>, } impl Section { @@ -2554,19 +2577,11 @@ impl Section { /// When the section file was last reached, why it could not be since, and what waits /// for it. `notify` runs when it is first reached and when an error comes or goes. pub fn sync_status(&self) -> Result { - let queued = self.replica.recovery_summary()?.queued_edits; - let observed = self - .observed - .lock() - .map_err(|_| io::Error::other("Synchronization observer panicked"))?; - Ok(SyncStatus { - synced: self.worker.as_ref().and_then(SyncWorker::synced), - error: observed - .as_ref() - .and_then(Option::as_ref) - .map(|error| io::Error::new(error.kind(), error.to_string())), - queued, - }) + sync_status( + &self.replica, + self.worker.as_ref().and_then(SyncWorker::synced), + &self.observed, + ) } /// The replica, which other threads may read pages from while the section is open. diff --git a/crates/notebook/src/worker.rs b/crates/notebook/src/worker.rs index f978c9eb1b6021f2932456d1aa504fd5f34dc2c0..f9526500b75e3b8e9b08a02c25c098934a7186fc 100644 --- a/crates/notebook/src/worker.rs +++ b/crates/notebook/src/worker.rs @@ -14,7 +14,7 @@ pub(super) struct Signal { pub(crate) stopped: AtomicBool, pub(crate) offline: AtomicBool, /// FILETIME of the last step or poll that reached the remote; 0 before one has. - synced: AtomicU64, + pub(crate) synced: AtomicU64, /// Asked for by `wake`: steps run while offline until one leaves nothing to do. pub(crate) requested: AtomicBool, /// A watch on the file's folder wakes the worker when the file changes, so an idle worker