diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index de904d0..ae0e043 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -1225,6 +1225,12 @@ pub struct JobItem { /// Steps known to need doing, or `0` when unknown (indeterminate progress). /// May *grow* mid-job as more work is discovered. pub total: u64, + /// Housekeeping that changes nothing in the user's files: a mirror folder + /// scan that finds nothing to do yet, an index rebuild, a cache check. Front + /// ends may show it, but should not report the drive as busy over it. Absent + /// from older daemons, which never set it. + #[serde(default)] + pub background: bool, } /// One in-flight transfer in a [`Response::Transfers`] snapshot. diff --git a/crates/pdfs-fuse/src/background.rs b/crates/pdfs-fuse/src/background.rs index 00772e6..7f14d5d 100644 --- a/crates/pdfs-fuse/src/background.rs +++ b/crates/pdfs-fuse/src/background.rs @@ -710,7 +710,7 @@ fn scan_local_once( // The walk has no idea how many files it will find, so the job counts what it // has seen and stays indeterminate. - let job = transfers.begin_job("Indexing this computer"); + let job = transfers.begin_background_job("Indexing this computer"); job.detail("Scanning your files"); let walked = localindex::scan(&[home], &excludes, |batch| { if let Err(e) = db.local_upsert_batch(generation, &batch) { diff --git a/crates/pdfs-fuse/src/devices.rs b/crates/pdfs-fuse/src/devices.rs index a432b34..476b4e0 100644 --- a/crates/pdfs-fuse/src/devices.rs +++ b/crates/pdfs-fuse/src/devices.rs @@ -657,12 +657,16 @@ impl Core { } else { p.total.max(p.done) as u64 }, + // Most passes find nothing; only one that queues work is + // the drive syncing. + background: true, }, SyncPhase::Applying => JobItem { title: format!("Syncing {folder}"), detail: p.current.clone(), done: p.done as u64, total: p.total.max(p.done) as u64, + background: false, }, }); } diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index d2a75b7..ff95fe1 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -1034,7 +1034,9 @@ impl Core { let core = self.clone(); std::thread::spawn(move || { - let job = core.transfers.begin_job("Checking the cache database"); + let job = core + .transfers + .begin_background_job("Checking the cache database"); job.detail("Verifying every page"); let problems = match core.db.integrity_check() { Ok(problems) => problems, diff --git a/crates/pdfs-fuse/src/sync.rs b/crates/pdfs-fuse/src/sync.rs index ae8d21e..abe74ba 100644 --- a/crates/pdfs-fuse/src/sync.rs +++ b/crates/pdfs-fuse/src/sync.rs @@ -50,7 +50,7 @@ mod planner; use engine::SYNC_CONCURRENCY; pub(crate) use engine::{SyncMsg, spawn}; #[cfg(test)] -use engine::{classify, settle_with}; +use engine::{classify, event_target, settle_with}; #[cfg(test)] use local_scan::is_write_mode; use local_scan::open_for_write_set; @@ -2097,6 +2097,32 @@ mod tests { (ids, started.elapsed()) } + #[test] + fn a_change_in_one_mirror_folder_reconciles_only_that_folder() { + let watched = vec![ + (PathBuf::from("/home/me/Docs"), 1), + (PathBuf::from("/home/me/Photos"), 2), + (PathBuf::from("/home/me/Photos/Raw"), 3), + ]; + let target = |paths: &[&str]| { + let paths: Vec = paths.iter().map(PathBuf::from).collect(); + match event_target(&watched, &paths) { + SyncMsg::Reconcile(id) => Some(id), + SyncMsg::ReconcileAll => None, + _ => panic!("unexpected message"), + } + }; + assert_eq!(target(&["/home/me/Docs/a.txt"]), Some(1)); + // A rename within one folder. + assert_eq!(target(&["/home/me/Docs/a", "/home/me/Docs/b"]), Some(1)); + // The deepest root wins. + assert_eq!(target(&["/home/me/Photos/Raw/x.cr2"]), Some(3)); + // Across folders, outside them all, or no path at all: walk everything. + assert_eq!(target(&["/home/me/Docs/a", "/home/me/Photos/a"]), None); + assert_eq!(target(&["/home/me/Docsx/a"]), None); + assert_eq!(target(&[]), None); + } + /// The bug this replaced a fixed sleep for: a save that keeps writing past /// the debounce used to be walked mid-write, uploading a torn snapshot as a /// real revision. The window has to restart on every event, so settling diff --git a/crates/pdfs-fuse/src/sync/engine.rs b/crates/pdfs-fuse/src/sync/engine.rs index 688aeb0..7755b5f 100644 --- a/crates/pdfs-fuse/src/sync/engine.rs +++ b/crates/pdfs-fuse/src/sync/engine.rs @@ -57,9 +57,10 @@ pub(crate) fn spawn(core: Core, rx: Receiver) -> Option) { // Paths the watcher currently covers, mapped to their folder id, so an event // path can be resolved back to the folder it belongs to. - let watched: Mutex> = Mutex::new(Vec::new()); + let watched: Arc>> = Arc::new(Mutex::new(Vec::new())); let tx_events = core.sync_tx.clone(); + let watched_events = watched.clone(); let mut watcher = match notify::recommended_watcher(move |res: notify::Result| { let Ok(event) = res else { return }; @@ -67,7 +68,8 @@ fn engine_loop(core: Core, rx: Receiver) { if matches!(event.kind, notify::EventKind::Access(_)) { return; } - let _ = tx_events.send(SyncMsg::ReconcileAll); + let msg = event_target(&watched_events.lock(), &event.paths); + let _ = tx_events.send(msg); }) { Ok(w) => w, Err(e) => { @@ -256,8 +258,31 @@ fn reconcile_all(core: &Core) { } } +/// The reconcile a watcher event asks for: the one mirror folder all its paths +/// lie in, so an edit in one folder does not walk every other one too. An event +/// that spans folders, or that no watched folder claims (a watch list a moment +/// out of date), falls back to every folder. +pub(super) fn event_target(watched: &[(PathBuf, i64)], paths: &[PathBuf]) -> SyncMsg { + let mut ids = paths.iter().map(|path| { + watched + .iter() + .filter(|(root, _)| path.starts_with(root)) + // The deepest root, should one mirror folder sit inside another. + .max_by_key(|(root, _)| root.components().count()) + .map(|(_, id)| *id) + }); + match ids.next() { + Some(Some(first)) if ids.all(|id| id == Some(first)) => SyncMsg::Reconcile(first), + _ => SyncMsg::ReconcileAll, + } +} + /// Bring the filesystem watches in line with the current mirror-folder set: /// watch newly-added folders, drop removed ones. +/// +/// The watch calls run on a copy, outside the lock: the watcher's event +/// callback takes the same lock, and a backend that hands a watch request to +/// its event thread and waits for the answer would deadlock against it. fn rewatch( core: &Core, watcher: &mut notify::RecommendedWatcher, @@ -274,7 +299,7 @@ fn rewatch( return; } }; - let mut have = watched.lock(); + let mut have = watched.lock().clone(); // Drop watches no longer wanted. have.retain(|(path, _)| { if want.iter().any(|(p, _)| p == path) { @@ -294,6 +319,7 @@ fn rewatch( Err(e) => warn!(path = %path.display(), error = %e, "sync: watch failed"), } } + *watched.lock() = have; } // ---- reconcile ------------------------------------------------------------ diff --git a/crates/pdfs-fuse/src/transfers.rs b/crates/pdfs-fuse/src/transfers.rs index 078f020..3e461b1 100644 --- a/crates/pdfs-fuse/src/transfers.rs +++ b/crates/pdfs-fuse/src/transfers.rs @@ -46,6 +46,8 @@ struct JobEntry { detail: Mutex, done: AtomicU64, total: AtomicU64, + /// Housekeeping that moves nothing; see [`JobItem::background`]. + background: bool, } /// The set of transfers and jobs currently in flight. Cloned `Arc`-style across @@ -160,12 +162,25 @@ impl TransferRegistry { /// that carries its progress and deregisters the job when dropped. The job /// starts indeterminate; give it a `total` once one is known. pub fn begin_job(self: &Arc, title: impl Into) -> JobGuard { + self.register_job(title.into(), false) + } + + /// [`TransferRegistry::begin_job`] for housekeeping the user did not ask + /// for and that changes nothing in their files, such as an index rebuild or + /// a cache check. It is reported, but front ends don't call the drive busy + /// over it. + pub fn begin_background_job(self: &Arc, title: impl Into) -> JobGuard { + self.register_job(title.into(), true) + } + + fn register_job(self: &Arc, title: String, background: bool) -> JobGuard { let id = self.next.fetch_add(1, Ordering::Relaxed); let entry = Arc::new(JobEntry { - title: title.into(), + title, detail: Mutex::new(String::new()), done: AtomicU64::new(0), total: AtomicU64::new(0), + background, }); self.jobs.lock().insert(id, entry.clone()); JobGuard { @@ -191,6 +206,7 @@ impl TransferRegistry { detail: e.detail.lock().clone(), done: e.done.load(Ordering::Relaxed), total: e.total.load(Ordering::Relaxed), + background: e.background, } }) .collect() diff --git a/crates/pdfs-gui/src/main.rs b/crates/pdfs-gui/src/main.rs index 27ae2d0..7e6cd8b 100644 --- a/crates/pdfs-gui/src/main.rs +++ b/crates/pdfs-gui/src/main.rs @@ -88,8 +88,9 @@ struct DriveState { /// Summarise a work snapshot into one menu line, or empty when idle. A single /// transfer names the file and its percentage; several collapse to counts so the /// menu stays one line regardless of queue depth. With nothing moving bytes, a -/// running job speaks for itself — a scan or an index rebuild is still "busy", -/// and the tray saying nothing there reads as "finished". +/// running job speaks for itself — a scan or an index rebuild is still work +/// going on, and the menu saying nothing there reads as "finished". The icon is +/// [`busy`]'s call, not this line's. fn sync_line(items: &[TransferItem], jobs: &[JobItem]) -> String { match items { [] => match jobs.first() { @@ -143,12 +144,12 @@ struct DriveTray { fn poll_state(socket: &Path, default_mountpoint: &Path) -> DriveState { match send(socket, &Request::Status) { Ok(status @ Response::Status { .. }) => { - // Same daemon is up, so a cheap follow-up request gives the sync line. - let sync = match send(socket, &Request::GetQueueStatus) { - Ok(Response::Transfers { items, jobs }) => sync_line(&items, &jobs), - _ => String::new(), + // Same daemon is up, so a cheap follow-up request says what moves. + let (items, jobs) = match send(socket, &Request::GetQueueStatus) { + Ok(Response::Transfers { items, jobs }) => (items, jobs), + _ => (Vec::new(), Vec::new()), }; - status_state(&status, sync, default_mountpoint) + status_state(&status, &items, &jobs, default_mountpoint) } // Socket answered but with something unexpected — treat as up but odd. Ok(_) => unexpected_reply(default_mountpoint), @@ -217,9 +218,14 @@ fn unexpected_reply(default_mountpoint: &Path) -> DriveState { } } -/// The state a running daemon's [`Response::Status`] describes, with `sync` -/// saying what it is moving right now. -fn status_state(status: &Response, sync: String, default_mountpoint: &Path) -> DriveState { +/// The state a running daemon's [`Response::Status`] describes, with `items` +/// and `jobs` saying what it is working on right now. +fn status_state( + status: &Response, + items: &[TransferItem], + jobs: &[JobItem], + default_mountpoint: &Path, +) -> DriveState { let Response::Status { mountpoint, pinned, @@ -239,7 +245,7 @@ fn status_state(status: &Response, sync: String, default_mountpoint: &Path) -> D paused, failing_ops, online, - queued.is_some() || !sync.is_empty(), + queued.is_some() || busy(items, jobs), ); DriveState { line: match (online, queued) { @@ -278,11 +284,19 @@ fn status_state(status: &Response, sync: String, default_mountpoint: &Path) -> D mounted: true, paused, mountpoint: PathBuf::from(mountpoint), - sync, + sync: sync_line(items, jobs), failing: failing_ops, } } +/// Whether the daemon is changing something: bytes moving, or a job the user +/// is waiting on. A mirror folder's routine scan and the like are background +/// jobs; they still get the menu line, but the icon spinning every two minutes +/// for a scan that finds nothing reads as constant syncing. +fn busy(items: &[TransferItem], jobs: &[JobItem]) -> bool { + !items.is_empty() || jobs.iter().any(|j| !j.background) +} + /// The phase a running mount is in. A pause is the user's own doing and says /// so first; failures outrank everything else the daemon reports. fn phase_of(paused: bool, failing: u64, online: bool, busy: bool) -> Phase { @@ -547,11 +561,7 @@ fn main() -> Result<()> { heard.take(feed); } let st = match &heard.status { - Some(status) => status_state( - status, - sync_line(&heard.items, &heard.jobs), - &default_mountpoint, - ), + Some(status) => status_state(status, &heard.items, &heard.jobs, &default_mountpoint), // Lost the daemon: ask once, which describes the sign-in instead. None => poll_state(&socket, &default_mountpoint), }; @@ -596,6 +606,39 @@ mod tests { assert_eq!(phase_of(false, 0, true, false), Phase::Synced); } + fn job(title: &str, background: bool) -> JobItem { + JobItem { + title: title.to_string(), + detail: String::new(), + done: 0, + total: 0, + background, + } + } + + #[test] + fn a_routine_scan_is_not_busy_but_a_sync_is() { + assert!(!busy(&[], &[])); + assert!(!busy(&[], &[job("Checking Documents", true)])); + assert!(busy( + &[], + &[ + job("Checking Documents", true), + job("Syncing Photos", false) + ] + )); + assert!(busy(&[], &[job("Uploading files", false)])); + } + + #[test] + fn a_job_from_an_older_daemon_counts_as_busy() { + let job: JobItem = serde_json::from_str( + r#"{"title":"Checking Documents","detail":"","done":0,"total":0}"#, + ) + .unwrap(); + assert!(!job.background); + } + #[test] fn the_issues_item_counts_in_the_singular_and_plural() { assert_eq!(issues_label(1), "1 issue — View");