From 5f50b6a9b5be1735c89716498f92a9c9b0134010 Mon Sep 17 00:00:00 2001 From: clover caruso Date: Sat, 10 Oct 2026 20:48:39 -0700 Subject: [PATCH] fix: report live sync status for held sections Derive held section status from weak references to its worker, replica, and observer after releasing the background watch lock. Share status derivation with Section, counting queued edits without scanning recovery assets. Cover offline queue publication, live remote error recovery, and cache release while the background watch retains its pre-hold status. Fixes #117 Assisted-by: gpt-6.1-sol --- crates/notebook/src/background.rs | 174 +++++++++++++++++++++++++++--- crates/notebook/src/session.rs | 45 +++++--- crates/notebook/src/worker.rs | 2 +- 3 files changed, 193 insertions(+), 28 deletions(-) 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 -- 2.54.0