Merge pull request #56 from narrrl/fix/sync-engine-containment

fix(fuse,core,cli): keep a panicking sync pass from ending the engine

crates/pdfs-cli/src/main.rs

@@ -3105,6 +3105,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)? {
@@ -3140,6 +3158,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 {

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<SyncEngineState>,
+}
+
+/// 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<i64>,
+    /// 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`].

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()),
         }
     }
 

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<Mutex<HashMap<i64, SyncProgress>>>,
+    /// Whether the folder-sync engine is alive, and which pass it is on.
+    sync_engine: Arc<sync::EngineHealth>,
     /// 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<sync::SyncMsg>,
@@ -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

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())),

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)));
+    }
 }

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<i64> {
 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).

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<SyncMsg>) -> Option<std::thread::Jo
     }
 }
 
+/// Whether the engine is alive and which pass it is on, for
+/// [`Core::diagnostics`] and the supervisor.
+///
+/// A panic in a pass used to end the engine thread. Every later add, sync-now,
+/// file event and poll then went into a channel nobody read, the folder stayed
+/// `syncing`, and nothing said so until a restart (`docs/BUGS.md` B259).
+#[derive(Default)]
+pub(crate) struct EngineHealth {
+    running: AtomicBool,
+    pass: Mutex<Option<PassClock>>,
+    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<SyncMsg>) {
+    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<Mutex<Vec<(PathBuf, i64)>>> = Arc::new(Mutex::new(Vec::new()));
@@ -120,9 +242,7 @@ fn engine_loop(core: Core, rx: Receiver<SyncMsg>) {
                 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<SyncMsg>) {
                 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);
     }
 }
 

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 |
 | [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 |
 | [B67](#b67--fresh-state-restore-can-silently-select-a-new-device-high-22) | Fresh-state restore can silently select a new device (HIGH-22) | An ambiguous device after a loss of local state is not asked about |
@@ -95,6 +94,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.
 
 ---