From c64bb23e3c8e55a5a3a51dd29ac825f964c062fb Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Sat, 31 Oct 2026 10:14:23 +0100 Subject: [PATCH] fix(fuse,core,cli): keep a panicking sync pass from ending the engine One thread runs every synced folder's passes. A panic in a pass ended it, and every later add, sync-now, file event and poll went nowhere without a word until a restart, which fits issue #53 (B259). Each pass now runs under catch_unwind; a panic sets the folder to error and the next poll tries again. Diagnostics reports the engine's state, the supervisor warns when it has ended or a pass has gone quiet, and first passes log at info. --- crates/pdfs-cli/src/main.rs | 21 +++++ crates/pdfs-core/src/control.rs | 23 +++++ crates/pdfs-fuse/src/diagnostics.rs | 1 + crates/pdfs-fuse/src/lib.rs | 3 + crates/pdfs-fuse/src/mount.rs | 1 + crates/pdfs-fuse/src/supervisor.rs | 68 +++++++++++++- crates/pdfs-fuse/src/sync.rs | 73 ++++++++++++++- crates/pdfs-fuse/src/sync/engine.rs | 132 ++++++++++++++++++++++++++-- docs/BUGS.md | 26 ++++-- 9 files changed, 334 insertions(+), 14 deletions(-) diff --git a/crates/pdfs-cli/src/main.rs b/crates/pdfs-cli/src/main.rs index 75f6e3b..b0b7e49 100644 --- a/crates/pdfs-cli/src/main.rs +++ b/crates/pdfs-cli/src/main.rs @@ -3096,6 +3096,24 @@ fn human_secs(secs: u64) -> String { } } +/// The sync engine's line in `pdfs diagnostics`: whether it runs, and which +/// folder's pass it is on. +fn sync_engine_line(sync: &pdfs_core::control::SyncEngineState) -> String { + let mut line = match (sync.running, sync.pass_folder) { + (false, _) => "STOPPED; synced folders do not sync until the daemon restarts".to_string(), + (true, None) => "idle".to_string(), + (true, Some(folder)) => format!( + "pass on folder {folder} for {}, last progress {} ago", + human_secs(sync.pass_age_secs), + human_secs(sync.pass_idle_secs), + ), + }; + if sync.panics > 0 { + line.push_str(&format!("; {} pass(es) panicked", sync.panics)); + } + line +} + fn cmd_diagnostics() -> Result<()> { let response = control_request(CtlRequest::Diagnostics)?; if emit_json(&response)? { @@ -3131,6 +3149,9 @@ fn cmd_diagnostics() -> Result<()> { human_secs(worker.age_secs) ); } + if let Some(sync) = &d.sync { + println!("sync engine: {}", sync_engine_line(sync)); + } if d.inflight.is_empty() { println!("in-flight requests: none"); } else { diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index c111146..3b7c590 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -1980,6 +1980,29 @@ pub struct Diagnostics { pub control_handler_limit: usize, /// Queued mutations not yet drained. pub pending_ops: u64, + /// The folder-sync engine. `None` from a daemon that predates the field. + #[serde(default)] + pub sync: Option, +} + +/// What the folder-sync engine is doing, for [`Diagnostics`]. +/// +/// The engine is one thread that runs every synced folder's passes in turn. +/// Dead or stuck, it looked exactly like an idle one: workers idle, nothing +/// queued, nothing in the journal (`docs/BUGS.md` B259). +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct SyncEngineState { + /// Whether the engine thread is running. Once it has ended, nothing syncs + /// until the daemon restarts. + pub running: bool, + /// The synced folder whose pass is running, if one is. + pub pass_folder: Option, + /// How long that pass has been running. + pub pass_age_secs: u64, + /// How long since that pass last reported progress. + pub pass_idle_secs: u64, + /// Passes that ended in a panic since the daemon started. + pub panics: u64, } /// The daemon's reply to a [`Request`]. diff --git a/crates/pdfs-fuse/src/diagnostics.rs b/crates/pdfs-fuse/src/diagnostics.rs index 5cde5df..f775043 100644 --- a/crates/pdfs-fuse/src/diagnostics.rs +++ b/crates/pdfs-fuse/src/diagnostics.rs @@ -163,6 +163,7 @@ impl Core { control_handlers: super::control::active_handlers(), control_handler_limit: super::control::handler_limit(), pending_ops: self.pending_ops_quickly(), + sync: Some(self.sync_engine.report()), } } diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index d9bbb0e..bb7f004 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -728,6 +728,8 @@ struct Core { /// can say what a pass is doing rather than just "syncing". An entry exists /// only while that folder's reconcile pass is running. sync_progress: Arc>>, + /// Whether the folder-sync engine is alive, and which pass it is on. + sync_engine: Arc, /// Channel to the folder-sync engine: nudges it to /// reconcile a folder, reconcile everything, or re-scan its watch set. sync_tx: std::sync::mpsc::Sender, @@ -5989,6 +5991,7 @@ impl Core { if let Some(p) = self.sync_progress.lock().get_mut(&folder_id) { f(p); } + self.sync_engine.moved(); } /// Set the number of items the scan expects to check, from the size of the diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 1aeccdc..f60a1e3 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -507,6 +507,7 @@ pub(crate) fn mount_with( events: Arc::new(Default::default()), indexing: Arc::new(AtomicBool::new(false)), sync_progress: Arc::new(Mutex::new(HashMap::new())), + sync_engine: Arc::default(), sync_tx, mounts: Arc::new(Mutex::new(SecondaryMountRegistry::default())), sync_locks: Arc::new(Mutex::new(HashMap::new())), diff --git a/crates/pdfs-fuse/src/supervisor.rs b/crates/pdfs-fuse/src/supervisor.rs index c7cfdf3..6e54f87 100644 --- a/crates/pdfs-fuse/src/supervisor.rs +++ b/crates/pdfs-fuse/src/supervisor.rs @@ -36,6 +36,11 @@ const TICK: Duration = Duration::from_secs(15); /// would train everyone to ignore the real one. const STALL_AFTER: Duration = Duration::from_secs(120); +/// How long a sync pass may go without reporting progress before it is called +/// out. Much longer than [`STALL_AFTER`]: one large file in a pass reports +/// nothing until its whole transfer is done. +const SYNC_STALL_AFTER: Duration = Duration::from_secs(900); + /// How often resident size goes into the log. const RSS_EVERY: Duration = Duration::from_secs(300); @@ -137,6 +142,26 @@ fn report_stalls(report: &pdfs_core::control::Diagnostics, last_completed: (u64, "the transfer lane has not completed a job since the last check" ); } + // The engine is one thread for every synced folder, so a dead or stuck one + // stops all of them while the worker pool looks idle (B259). + if let Some(sync) = &report.sync { + if !sync.running { + stalled = true; + warn!( + "the sync engine is not running; synced folders do not sync until the daemon restarts" + ); + } else if let Some(folder) = sync.pass_folder + && sync.pass_idle_secs >= SYNC_STALL_AFTER.as_secs() + { + stalled = true; + warn!( + folder, + age_secs = sync.pass_age_secs, + idle_secs = sync.pass_idle_secs, + "a sync pass has not reported progress for a long time" + ); + } + } stalled } @@ -169,7 +194,7 @@ fn probe(control_socket: &Path) -> bool { #[cfg(test)] mod tests { use super::*; - use pdfs_core::control::{Diagnostics, InflightRequest, WorkerState}; + use pdfs_core::control::{Diagnostics, InflightRequest, SyncEngineState, WorkerState}; fn busy_worker(age_secs: u64) -> WorkerState { WorkerState { @@ -226,4 +251,45 @@ mod tests { assert!(!report_stalls(&report, (0, 30))); assert!(report_stalls(&report, (0, 40))); } + + fn with_sync(sync: SyncEngineState) -> Diagnostics { + Diagnostics { + sync: Some(sync), + ..Default::default() + } + } + + /// A sync engine that has ended stops every synced folder while the + /// workers look idle, which is what issue #53 looked like (B259). + #[test] + fn a_stopped_sync_engine_is_a_stall() { + assert!(report_stalls( + &with_sync(SyncEngineState::default()), + (0, 0) + )); + } + + /// A pass that keeps reporting progress is working, however long it runs; + /// one that has gone quiet past the threshold is not. + #[test] + fn only_a_quiet_sync_pass_is_a_stall() { + let busy = SyncEngineState { + running: true, + pass_folder: Some(3), + pass_age_secs: SYNC_STALL_AFTER.as_secs() * 4, + pass_idle_secs: 5, + panics: 0, + }; + assert!(!report_stalls(&with_sync(busy.clone()), (0, 0))); + let quiet = SyncEngineState { + pass_idle_secs: SYNC_STALL_AFTER.as_secs(), + ..busy + }; + assert!(report_stalls(&with_sync(quiet), (0, 0))); + let idle = SyncEngineState { + running: true, + ..Default::default() + }; + assert!(!report_stalls(&with_sync(idle), (0, 0))); + } } diff --git a/crates/pdfs-fuse/src/sync.rs b/crates/pdfs-fuse/src/sync.rs index 864e522..c698021 100644 --- a/crates/pdfs-fuse/src/sync.rs +++ b/crates/pdfs-fuse/src/sync.rs @@ -48,8 +48,10 @@ mod engine; mod local_scan; mod model; mod planner; +#[cfg(test)] +use engine::Running; pub(crate) use engine::SYNC_CONCURRENCY; -pub(crate) use engine::{SyncMsg, spawn}; +pub(crate) use engine::{EngineHealth, SyncMsg, spawn}; #[cfg(test)] use engine::{classify, event_target, settle_with}; #[cfg(test)] @@ -67,6 +69,11 @@ use planner::{ plan_file, rel_to_path, remote_sig, remote_subtree_changed, revision_moved, tracked_baseline, unchanged_remote_size, }; + +/// A pass that took at least this long is logged at info even when it changed +/// nothing, so the journal says where a slow engine spends its time. +const SLOW_PASS: Duration = Duration::from_secs(30); + /// What a [`Pending`] op did, so the reconcile can update shared state on the /// engine thread (never inside a task). enum Applied { @@ -154,6 +161,14 @@ impl Core { // The cost is one small file read against a pass that is about to walk the // whole tree. let rules = self.ignore_rules(&local_root); + // A first pass is the one a user waits on after adding or restoring a + // folder, so it is said at info; the poll's passes stay at debug. + if folder.last_sync == 0 { + info!(id = folder.id, path = %folder.local_path, "sync: first pass starting"); + } else { + debug!(id = folder.id, "sync: pass starting"); + } + let started = Instant::now(); self.progress_begin(folder.id); let result = if push_only { self.push_pass(folder.id, &local_root, &remote_root, &rules) @@ -195,6 +210,18 @@ impl Core { "idle" }; let _ = self.db.sync_folder_set_state(folder.id, state, now_secs()); + let elapsed = started.elapsed(); + if folder.last_sync == 0 || !outcome.is_empty() || elapsed >= SLOW_PASS { + info!( + id = folder.id, + state, + elapsed_ms = elapsed.as_millis() as u64, + summary = %outcome.summary(), + "sync: pass finished" + ); + } else { + debug!(id = folder.id, state, "sync: pass finished"); + } self.report_unscanned(folder.id, name, &outcome.unscanned); // Summarise a pass that actually moved something. A poll over an // unchanged folder does nothing and says nothing — otherwise the @@ -2438,6 +2465,50 @@ fn system_mtime_ns(meta: &std::fs::Metadata) -> Option { mod tests { use super::*; + /// A panic in a pass ends that pass and is counted; the engine is not + /// left believing the pass still runs (`docs/BUGS.md` B259). + #[test] + fn a_panicking_pass_is_contained_and_counted() { + let health = EngineHealth::default(); + assert!(!health.contain(7, || panic!("a pass that panics"))); + let report = health.report(); + assert_eq!(report.panics, 1); + assert_eq!(report.pass_folder, None); + assert!(health.contain(7, || {})); + assert_eq!(health.report().panics, 1); + } + + /// While a pass runs, the report names its folder, which is what tells a + /// long pass apart from an idle engine. + #[test] + fn the_running_pass_is_reported() { + let health = EngineHealth::default(); + let mut seen = None; + health.contain(7, || seen = health.report().pass_folder); + assert_eq!(seen, Some(7)); + } + + /// The engine reads as running only while its thread lives, including + /// when the thread ends in a panic. + #[test] + fn the_engine_reads_as_stopped_once_its_thread_ends() { + let health = Arc::new(EngineHealth::default()); + let thread_health = health.clone(); + let (started_tx, started_rx) = std::sync::mpsc::channel(); + let (end_tx, end_rx) = std::sync::mpsc::channel::<()>(); + let thread = std::thread::spawn(move || { + let _running = Running::start(&thread_health); + started_tx.send(()).unwrap(); + let _ = end_rx.recv(); + panic!("the engine thread panics"); + }); + started_rx.recv().unwrap(); + assert!(health.report().running); + end_tx.send(()).unwrap(); + assert!(thread.join().is_err()); + assert!(!health.report().running); + } + /// A removal of something already gone has produced the state the sync /// engine wanted, so its baseline is dropped rather than retried forever. /// Anything else stays an error and keeps the baseline (`docs/BUGS.md` B42). diff --git a/crates/pdfs-fuse/src/sync/engine.rs b/crates/pdfs-fuse/src/sync/engine.rs index 5664392..7020cb4 100644 --- a/crates/pdfs-fuse/src/sync/engine.rs +++ b/crates/pdfs-fuse/src/sync/engine.rs @@ -1,5 +1,11 @@ //! Filesystem watcher, polling, and debounce control plane. +use std::panic::AssertUnwindSafe; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + +use pdfs_core::control::SyncEngineState; +use tracing::error; + use super::*; /// How long the watcher must go quiet before a burst counts as finished. @@ -53,8 +59,124 @@ pub(crate) fn spawn(core: Core, rx: Receiver) -> Option>, + panics: AtomicU64, +} + +#[derive(Clone, Copy)] +struct PassClock { + folder: i64, + started: Instant, + moved: Instant, +} + +impl EngineHealth { + /// Note that the running pass reported progress. + pub(crate) fn moved(&self) { + if let Some(pass) = self.pass.lock().as_mut() { + pass.moved = Instant::now(); + } + } + + /// The engine's state now. `try_lock`, like the rest of the diagnostics: + /// a report must not wait on the thing it reports about. + pub(crate) fn report(&self) -> SyncEngineState { + let pass = self.pass.try_lock().and_then(|pass| *pass); + SyncEngineState { + running: self.running.load(Ordering::Relaxed), + pass_folder: pass.map(|p| p.folder), + pass_age_secs: pass.map_or(0, |p| p.started.elapsed().as_secs()), + pass_idle_secs: pass.map_or(0, |p| p.moved.elapsed().as_secs()), + panics: self.panics.load(Ordering::Relaxed), + } + } + + /// Run `pass` over `folder` as the current pass. Returns `false` when it + /// panicked: the panic ends that pass, not the engine. + pub(super) fn contain(&self, folder: i64, pass: impl FnOnce()) -> bool { + let now = Instant::now(); + *self.pass.lock() = Some(PassClock { + folder, + started: now, + moved: now, + }); + let finished = std::panic::catch_unwind(AssertUnwindSafe(pass)).is_ok(); + *self.pass.lock() = None; + if !finished { + self.panics.fetch_add(1, Ordering::Relaxed); + } + finished + } +} + +/// Marks the engine running for as long as it lives. +pub(super) struct Running<'a>(&'a EngineHealth); + +impl<'a> Running<'a> { + pub(super) fn start(health: &'a EngineHealth) -> Self { + health.running.store(true, Ordering::Relaxed); + Self(health) + } +} + +impl Drop for Running<'_> { + fn drop(&mut self) { + self.0.running.store(false, Ordering::Relaxed); + if std::thread::panicking() { + error!("sync engine stopped by a panic; nothing syncs until the daemon restarts"); + } + } +} + +/// One folder's pass, contained: a panic in it sets the folder to `error` and +/// leaves the engine running for the next one. +fn run_pass(core: &Core, folder: &StoredSyncFolder) { + if core + .sync_engine + .contain(folder.id, || core.reconcile_folder(folder)) + { + return; + } + core.progress_end(folder.id); + let _ = core + .db + .sync_folder_set_state(folder.id, "error", folder.last_sync); + error!( + id = folder.id, + path = %folder.local_path, + "sync: pass panicked; the folder is set to error and tried again on the next poll" + ); + core.log_activity( + ActivityKind::Sync, + format!("sync failed for {}", base_name(&folder.local_path)), + "an internal error stopped the pass; it is tried again on the next poll", + false, + ); +} + +/// The pass a `Reconcile(id)` asked for. A folder removed since, or one the +/// database cannot read, is said so rather than dropped in silence. +fn run_pass_by_id(core: &Core, id: i64) { + match core.db.sync_folder_get(id) { + Ok(Some(folder)) => run_pass(core, &folder), + Ok(None) => debug!(id, "sync: folder removed before its pass; skipping"), + Err(e) => warn!(id, error = ?e, "sync: cannot read folder; skipping its pass"), + } +} + /// The engine control loop. Single-threaded so reconcile passes never overlap. fn engine_loop(core: Core, rx: Receiver) { + let health = core.sync_engine.clone(); + let _running = Running::start(&health); // 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: Arc>> = Arc::new(Mutex::new(Vec::new())); @@ -120,9 +242,7 @@ fn engine_loop(core: Core, rx: Receiver) { if core.sync_paused() { break; } - if let Ok(Some(folder)) = core.db.sync_folder_get(id) { - core.reconcile_folder(&folder); - } + run_pass_by_id(&core, id); } } } @@ -159,9 +279,7 @@ fn poll_only_loop(core: &Core, rx: Receiver) { if core.sync_paused() { break; } - if let Ok(Some(folder)) = core.db.sync_folder_get(id) { - core.reconcile_folder(&folder); - } + run_pass_by_id(core, id); } } } @@ -254,7 +372,7 @@ fn reconcile_all(core: &Core) { } }; for folder in folders { - core.reconcile_folder(&folder); + run_pass(core, &folder); } } diff --git a/docs/BUGS.md b/docs/BUGS.md index 319b4ca..2103de3 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -82,7 +82,6 @@ original wording. | [B262](#b262--watching-a-large-synced-folder-holds-up-the-sync-engine-and-a-failed-watch-is-never-retried) | Watching a large synced folder holds up the sync engine, and a failed watch is never retried | Not fixed | | [B261](#b261--a-folder-that-has-never-synced-reads-up-to-date) | A folder that has never synced reads "up to date" | Not fixed | | [B260](#b260--the-poll-pushes-a-new-folders-first-sync-behind-every-other-folder) | The poll pushes a new folder's first sync behind every other folder | Not fixed | -| [B259](#b259--the-sync-engine-can-die-or-hang-and-nothing-shows-it-until-a-restart) | The sync engine can die or hang, and nothing shows it until a restart | Not fixed; the trigger in #53 is not known | | [B258](#b258--journalctl--p-warning-shows-nothing-whatever-the-daemon-logged) | `journalctl -p warning` shows nothing, whatever the daemon logged | Not fixed | | [B134](#b134--a-file-deleted-online-while-the-drain-checks-its-write-is-kept-as-a-conflict-copy) | A file deleted online while the drain checks its write is kept as a conflict copy | Only with `local_first` off | | [B91](#b91--the-daemon-hung-overnight-and-nothing-could-say-what-it-was-waiting-on) | The daemon hung overnight, and nothing could say what it was waiting on | The cause is not diagnosed; the daemon now reports where it is stuck and is restarted | @@ -96,6 +95,7 @@ original wording. | Entry | Title | What is left | |---|---|---| +| [B259](#b259--the-sync-engine-can-die-or-hang-and-nothing-shows-it-until-a-restart) | The sync engine can die or hang, and nothing shows it until a restart | No pass deadline; an ended engine is reported, not restarted; the trigger in #53 is not known | | [B199](#b199--a-stopped-share-stays-on-the-shared-by-me-page-for-minutes) | A stopped share stays on the Shared by me page for minutes | The share itself is not deleted; the SDK has no call for it | | [B88](#b88--pdfs-pin-rename-move-delete-and-sharing-still-cannot-name-a-path-under-a-secondary-mount) | `pdfs pin`, `rename`, `move`, `delete` and sharing still cannot name a path under a secondary mount | `pin`, `unpin`, opening a file, thumbnails, uploads, sharing and versions | | [B72](#b72--pdfs-trash-never-returns) | `pdfs trash` never returns | The slowness of the first refresh is unmeasured | @@ -680,7 +680,8 @@ while a pass runs. ### B259 — The sync engine can die or hang, and nothing shows it until a restart -**Status:** Open +**Status:** Partly fixed. A hung pass is reported but not ended, and a panic outside a pass +still stops the engine; it is reported, not restarted. **Found:** 2026-10-31, issue #53: a newly added folder did not sync until the reporter restarted the app. All FUSE workers were idle, nothing was queued, and the warning-level journal was empty (B258). @@ -704,9 +705,24 @@ is why it helps. nothing about sync. - A database error when the engine reads a folder drops that folder's request without a log. -**Fix.** Not done. Catch a panic per pass, end its progress and set the folder to `error`. Have -the supervisor restart a finished engine thread. Log passes at info, and report the folder, -the pass age and the queue in `Diagnostics`. +**Fix.** +- Each pass runs under `catch_unwind` (`EngineHealth::contain` in + `crates/pdfs-fuse/src/sync/engine.rs`). A panic ends that pass only: its progress is cleared, + the folder is set to `error`, the activity feed says so, and the next poll tries again. +- `Diagnostics` carries the engine's state: running or not, the folder of the running pass, how + long it has run and how long since it last reported progress, and how many passes panicked. + `pdfs diagnostics` prints it as `sync engine: …`. +- The supervisor warns when the engine thread has ended, and when a pass has reported no + progress for 15 minutes. +- A first pass logs its start and end at info, as does any pass that changed something or + took 30 s or more. A request for a folder the database cannot read is logged. +- Not done: a deadline for a pass, and restarting an engine thread that has ended. The + channel's receiver ends with the thread, so a restart needs a new channel in every `Core` + clone. + +**Test:** `a_panicking_pass_is_contained_and_counted`, `the_running_pass_is_reported` and +`the_engine_reads_as_stopped_once_its_thread_ends` in `pdfs-fuse`; +`a_stopped_sync_engine_is_a_stall` and `only_a_quiet_sync_pass_is_a_stall` in the supervisor. ---