Merge branch 'fix/sync-scheduling-watches' into fix/never-synced-label

# Conflicts:
#	docs/BUGS.md

crates/pdfs-cli/src/journal.rs

@@ -0,0 +1,122 @@
+//! Log priorities for journald.
+//!
+//! journald files each line a service writes at the unit's default priority,
+//! info, unless the line opens with a syslog `<N>` prefix. Without one every
+//! WARN and ERROR the daemon logs is stored as info, and the
+//! `journalctl -p warning` that the troubleshooting guide asks for shows
+//! nothing (`docs/BUGS.md` B258).
+
+use std::fmt;
+use std::os::unix::fs::MetadataExt;
+
+use tracing::{Event, Level, Subscriber};
+use tracing_subscriber::fmt::format::Writer;
+use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields};
+use tracing_subscriber::registry::LookupSpan;
+
+/// Whether standard output is the journal stream systemd connected this
+/// process to.
+///
+/// `JOURNAL_STREAM` alone does not say so: a terminal started from a user unit
+/// inherits it, and every `pdfs` run in that terminal would print the
+/// prefixes. systemd sets it to the stream's `device:inode`, so it is compared
+/// against what fd 1 actually is.
+pub(crate) fn stdout_is_journal() -> bool {
+    let Some(stream) = std::env::var_os("JOURNAL_STREAM") else {
+        return false;
+    };
+    let Ok(stdout) = std::fs::metadata("/proc/self/fd/1") else {
+        return false;
+    };
+    stream
+        .to_str()
+        .is_some_and(|value| names_stream(value, stdout.dev(), stdout.ino()))
+}
+
+/// Whether a `JOURNAL_STREAM` value names the file `dev`:`ino`.
+fn names_stream(value: &str, dev: u64, ino: u64) -> bool {
+    value
+        .split_once(':')
+        .is_some_and(|(d, i)| d.parse() == Ok(dev) && i.parse() == Ok(ino))
+}
+
+/// The syslog priority journald should file a `level` line at.
+fn priority(level: &Level) -> u8 {
+    match *level {
+        Level::ERROR => 3,
+        Level::WARN => 4,
+        Level::INFO => 6,
+        Level::DEBUG | Level::TRACE => 7,
+    }
+}
+
+/// An event format that opens each line with its `<N>` priority, then writes
+/// the line as `F` would.
+pub(crate) struct Prioritized<F>(pub(crate) F);
+
+impl<S, N, F> FormatEvent<S, N> for Prioritized<F>
+where
+    S: Subscriber + for<'a> LookupSpan<'a>,
+    N: for<'a> FormatFields<'a> + 'static,
+    F: FormatEvent<S, N>,
+{
+    fn format_event(
+        &self,
+        ctx: &FmtContext<'_, S, N>,
+        mut writer: Writer<'_>,
+        event: &Event<'_>,
+    ) -> fmt::Result {
+        write!(writer, "<{}>", priority(event.metadata().level()))?;
+        self.0.format_event(ctx, writer, event)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io;
+    use std::sync::{Arc, Mutex};
+
+    #[derive(Clone)]
+    struct Buf(Arc<Mutex<Vec<u8>>>);
+
+    impl io::Write for Buf {
+        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+            self.0.lock().unwrap().extend_from_slice(bytes);
+            Ok(bytes.len())
+        }
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn each_line_opens_with_the_priority_of_its_level() {
+        let buf = Buf(Arc::new(Mutex::new(Vec::new())));
+        let writer = buf.clone();
+        let subscriber = tracing_subscriber::fmt()
+            .with_max_level(Level::DEBUG)
+            .with_ansi(false)
+            .with_writer(move || writer.clone())
+            .event_format(Prioritized(tracing_subscriber::fmt::format()))
+            .finish();
+        tracing::subscriber::with_default(subscriber, || {
+            tracing::error!("e");
+            tracing::warn!("w");
+            tracing::info!("i");
+            tracing::debug!("d");
+        });
+        let out = String::from_utf8(buf.0.lock().unwrap().clone()).unwrap();
+        let prefixes: Vec<&str> = out.lines().map(|line| &line[..3]).collect();
+        assert_eq!(prefixes, ["<3>", "<4>", "<6>", "<7>"]);
+    }
+
+    #[test]
+    fn journal_stream_must_name_the_same_file() {
+        assert!(names_stream("8:12345", 8, 12345));
+        assert!(!names_stream("8:12345", 8, 54321));
+        assert!(!names_stream("9:12345", 8, 12345));
+        assert!(!names_stream("8", 8, 12345));
+        assert!(!names_stream("", 8, 12345));
+    }
+}

crates/pdfs-cli/src/main.rs

@@ -20,6 +20,8 @@ use pdfs_core::control::{
 use pdfs_core::db::Db;
 use pdfs_core::service;
 
+mod journal;
+
 /// `println!` for this binary, shadowing std's, which panics when stdout is
 /// closed: `pdfs ls | head` ended in a "Broken pipe" panic once `head` had its
 /// lines. This one hands the failure to [`stdout_written`] instead.
@@ -879,15 +881,22 @@ fn cli_error(kind: ErrorKind, message: &str) -> String {
 }
 
 fn main() -> Result<()> {
-    tracing_subscriber::fmt()
+    let filter = tracing_subscriber::EnvFilter::try_from_default_env()
-        .with_env_filter(
-            tracing_subscriber::EnvFilter::try_from_default_env()
         // `pgp` logs a benign WARN when it re-serializes a Proton secret-key
         // packet with a longer (but equivalent) length encoding than the
         // original; drop it to error so normal runs stay quiet.
-                .unwrap_or_else(|_| "info,pgp=error".into()),
+        .unwrap_or_else(|_| "info,pgp=error".into());
-        )
+    if journal::stdout_is_journal() {
+        // journalctl colours lines by the priority the prefix gives them, so
+        // the escape codes would only be noise in the stored text.
+        tracing_subscriber::fmt()
+            .with_env_filter(filter)
+            .with_ansi(false)
+            .event_format(journal::Prioritized(tracing_subscriber::fmt::format()))
             .init();
+    } else {
+        tracing_subscriber::fmt().with_env_filter(filter).init();
+    }
 
     let cli = Cli::parse();
     JSON_OUTPUT.store(cli.json, std::sync::atomic::Ordering::Relaxed);
@@ -3096,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)? {
@@ -3131,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

@@ -49,7 +49,9 @@ mod local_scan;
 mod model;
 mod planner;
 pub(crate) use engine::SYNC_CONCURRENCY;
-pub(crate) use engine::{SyncMsg, spawn};
+pub(crate) use engine::{EngineHealth, SyncMsg, spawn};
+#[cfg(test)]
+use engine::{Running, burst_order, keep_wanted};
 #[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,72 @@ 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);
+    }
+
+    /// Folders named by id run before the rest of a `ReconcileAll` in the
+    /// same burst, so a folder just added does not wait behind a pass over
+    /// every other folder (`docs/BUGS.md` B260).
+    #[test]
+    fn named_folders_run_before_the_rest_of_a_burst() {
+        let ids: HashSet<i64> = [9, 4].into();
+        assert_eq!(burst_order(&ids, &[1, 4, 6, 9]), [4, 9, 1, 6]);
+        assert_eq!(burst_order(&HashSet::new(), &[1, 4]), [1, 4]);
+        // A folder the list no longer has still gets the pass it asked for.
+        assert_eq!(burst_order(&[12].into(), &[1]), [12, 1]);
+    }
+
+    /// A folder removed and added again keeps its watch under the new id;
+    /// a folder that is gone loses its watch (`docs/BUGS.md` B272).
+    #[test]
+    fn a_re_added_folder_keeps_its_watch_under_the_new_id() {
+        let mut have = vec![(PathBuf::from("/a"), 1), (PathBuf::from("/b"), 2)];
+        let dropped = keep_wanted(&mut have, &[(PathBuf::from("/a"), 7)]);
+        assert_eq!(have, [(PathBuf::from("/a"), 7)]);
+        assert_eq!(dropped, [PathBuf::from("/b")]);
+    }
+
+    /// 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.
@@ -23,6 +29,9 @@ const POLL_INTERVAL: Duration = Duration::from_secs(120);
 /// unusable done one-at-a-time; this bounds concurrency like the bulk-upload
 /// engine does.
 pub(crate) const SYNC_CONCURRENCY: usize = 8;
+/// How long after a failed watch the engine tries the watches again. Each try
+/// walks the folder's whole tree, so not on every burst.
+const REWATCH_RETRY: Duration = Duration::from_secs(600);
 
 /// A message to the sync engine's control thread.
 pub(crate) enum SyncMsg {
@@ -53,24 +62,151 @@ 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()));
 
     let tx_events = core.sync_tx.clone();
     let watched_events = watched.clone();
-    let mut watcher =
+    let handler = move |res: notify::Result<notify::Event>| {
-        match notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
+        let event = match res {
-            let Ok(event) = res else { return };
+            Ok(event) => event,
+            // Running out of inotify watches for a new subfolder, say: that
+            // subfolder's changes are then found by the poll only (B262).
+            Err(e) => {
+                warn!(error = %e, "sync: filesystem watcher error");
+                return;
+            }
+        };
         // Ignore pure metadata/access noise; content changes are what matter.
         if matches!(event.kind, notify::EventKind::Access(_)) {
             return;
         }
         let msg = event_target(&watched_events.lock(), &event.paths);
         let _ = tx_events.send(msg);
-        }) {
+    };
+    // The local walk skips symlinks, so following them here would only watch
+    // trees outside the folder, and run passes over it for changes made there.
+    let config = notify::Config::default().with_follow_symlinks(false);
+    let mut watcher = match notify::RecommendedWatcher::new(handler, config) {
         Ok(w) => w,
         Err(e) => {
             warn!(error = %e, "filesystem watcher unavailable; sync is poll-only");
@@ -96,8 +232,9 @@ fn engine_loop(core: Core, rx: Receiver<SyncMsg>) {
             .ok();
     }
 
-    rewatch(&core, &mut watcher, &watched);
+    let mut unwatched = rewatch(&core, &mut watcher, &watched);
-    reconcile_all(&core);
+    let mut last_rewatch = Instant::now();
+    run_burst(&core, &HashSet::new(), true);
 
     while let Ok(msg) = rx.recv() {
         let mut ids: HashSet<i64> = HashSet::new();
@@ -110,21 +247,16 @@ fn engine_loop(core: Core, rx: Receiver<SyncMsg>) {
             return;
         }
 
+        // A folder whose watch failed is tried again now and then, not only
+        // when the folder set changes.
+        if unwatched && last_rewatch.elapsed() >= REWATCH_RETRY {
+            do_rewatch = true;
+        }
         if do_rewatch {
-            rewatch(&core, &mut watcher, &watched);
+            unwatched = rewatch(&core, &mut watcher, &watched);
-        }
+            last_rewatch = Instant::now();
-        if all {
-            reconcile_all(&core);
-        } else {
-            for id in ids {
-                if core.sync_paused() {
-                    break;
-                }
-                if let Ok(Some(folder)) = core.db.sync_folder_get(id) {
-                    core.reconcile_folder(&folder);
-                }
-            }
         }
+        run_burst(&core, &ids, all);
     }
 }
 
@@ -141,7 +273,7 @@ fn poll_only_loop(core: &Core, rx: Receiver<SyncMsg>) {
             }
         });
     }
-    reconcile_all(core);
+    run_burst(core, &HashSet::new(), true);
     while let Ok(msg) = rx.recv() {
         let mut ids: HashSet<i64> = HashSet::new();
         let mut all = false;
@@ -152,19 +284,57 @@ fn poll_only_loop(core: &Core, rx: Receiver<SyncMsg>) {
             info!("sync engine stopping");
             return;
         }
-        if all {
+        run_burst(core, &ids, all);
-            reconcile_all(core);
+    }
+}
+
+/// Run the passes one burst asked for: the folders named by id, then, when the
+/// burst also holds a `ReconcileAll`, every other folder.
+///
+/// The named folders go first. A poll's `ReconcileAll` in the same burst used
+/// to take their place, and a folder just added, which has the highest id,
+/// waited behind a full pass over every other folder (`docs/BUGS.md` B260).
+///
+/// Skipped while sync is paused, and cut short when a pause lands between
+/// passes: the resume sends a fresh [`SyncMsg::ReconcileAll`], which walks
+/// everything the watcher reported meanwhile.
+fn run_burst(core: &Core, ids: &HashSet<i64>, all: bool) {
+    if core.sync_paused() {
+        debug!("sync: paused, skipping reconcile");
+        return;
+    }
+    let folders = if all {
+        match core.db.sync_folder_list() {
+            Ok(folders) => folders,
+            Err(e) => {
+                warn!(error = ?e, "sync: cannot list folders");
+                Vec::new()
+            }
+        }
     } else {
-            for id in ids {
+        Vec::new()
+    };
+    let listed: Vec<i64> = folders.iter().map(|f| f.id).collect();
+    for id in burst_order(ids, &listed) {
         if core.sync_paused() {
             break;
         }
-                if let Ok(Some(folder)) = core.db.sync_folder_get(id) {
+        // A folder named by id is read again: it may have been added or
-                    core.reconcile_folder(&folder);
+        // changed after the list was read.
-                }
+        match folders.iter().find(|f| f.id == id) {
+            Some(folder) if !ids.contains(&id) => run_pass(core, folder),
+            _ => run_pass_by_id(core, id),
         }
     }
 }
+
+/// The order of a burst's passes: the folders named by id, lowest first, then
+/// the rest of `listed` in its own order.
+pub(super) fn burst_order(ids: &HashSet<i64>, listed: &[i64]) -> Vec<i64> {
+    let mut order: Vec<i64> = ids.iter().copied().collect();
+    order.sort_unstable();
+    order.extend(listed.iter().copied().filter(|id| !ids.contains(id)));
+    order
 }
 
 /// Absorb the rest of an event burst, returning once the watcher has been quiet
@@ -238,26 +408,6 @@ pub(super) fn classify(
     true
 }
 
-/// Reconcile every mirror folder in turn. Skipped while sync is paused: the
-/// resume sends a fresh [`SyncMsg::ReconcileAll`], which walks everything the
-/// watcher reported meanwhile.
-fn reconcile_all(core: &Core) {
-    if core.sync_paused() {
-        debug!("sync: paused, skipping reconcile");
-        return;
-    }
-    let folders = match core.db.sync_folder_list() {
-        Ok(f) => f,
-        Err(e) => {
-            warn!(error = ?e, "sync: cannot list folders");
-            return;
-        }
-    };
-    for folder in folders {
-        core.reconcile_folder(&folder);
-    }
-}
-
 /// 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
@@ -278,7 +428,8 @@ pub(super) fn event_target(watched: &[(PathBuf, i64)], paths: &[PathBuf]) -> Syn
 }
 
 /// Bring the filesystem watches in line with the current mirror-folder set:
-/// watch newly-added folders, drop removed ones.
+/// watch newly-added folders, drop removed ones. Returns whether a folder is
+/// left unwatched because its watch failed.
 ///
 /// 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
@@ -287,7 +438,7 @@ fn rewatch(
     core: &Core,
     watcher: &mut notify::RecommendedWatcher,
     watched: &Mutex<Vec<(PathBuf, i64)>>,
-) {
+) -> bool {
     let want: Vec<(PathBuf, i64)> = match core.db.sync_folder_list() {
         Ok(f) => f
             .into_iter()
@@ -296,30 +447,60 @@ fn rewatch(
             .collect(),
         Err(e) => {
             warn!(error = ?e, "sync: cannot list folders to watch");
-            return;
+            return false;
         }
     };
     let mut have = watched.lock().clone();
     // Drop watches no longer wanted.
-    have.retain(|(path, _)| {
+    for path in keep_wanted(&mut have, &want) {
-        if want.iter().any(|(p, _)| p == path) {
+        let _ = watcher.unwatch(&path);
-            true
-        } else {
-            let _ = watcher.unwatch(path);
-            false
     }
-    });
     // Add newly wanted watches.
+    let mut failed = false;
     for (path, id) in &want {
         if have.iter().any(|(p, _)| p == path) {
             continue;
         }
         match watcher.watch(path, RecursiveMode::Recursive) {
             Ok(()) => have.push((path.clone(), *id)),
-            Err(e) => warn!(path = %path.display(), error = %e, "sync: watch failed"),
+            Err(e) => {
+                // A recursive watch that fails partway keeps the watches it had
+                // added. Their events would match no folder and run a pass over
+                // every one, and they hold watches the next folder needs (B262).
+                let _ = watcher.unwatch(path);
+                failed = true;
+                warn!(
+                    path = %path.display(),
+                    error = %e,
+                    "sync: watch failed; the poll finds this folder's changes until a retry \
+                     works. A low fs.inotify.max_user_watches is the usual cause"
+                );
+            }
         }
     }
     *watched.lock() = have;
+    failed
+}
+
+/// Keep the watches in `have` that `want` still asks for, and return the paths
+/// of the others, to unwatch.
+///
+/// A path is matched alone, but takes the id `want` gives it: a folder removed
+/// and added again in one burst keeps its watch, and its events go to the new
+/// row rather than to the removed one (`docs/BUGS.md` B272).
+pub(super) fn keep_wanted(have: &mut Vec<(PathBuf, i64)>, want: &[(PathBuf, i64)]) -> Vec<PathBuf> {
+    let mut dropped = Vec::new();
+    have.retain_mut(|(path, id)| match want.iter().find(|(p, _)| p == path) {
+        Some((_, wanted)) => {
+            *id = *wanted;
+            true
+        }
+        None => {
+            dropped.push(path.clone());
+            false
+        }
+    });
+    dropped
 }
 
 // ---- reconcile ------------------------------------------------------------

docs/BUGS.md

@@ -69,7 +69,6 @@ original wording.
 | Entry | Title | What is left |
 |---|---|---|
 | [B273](#b273--a-restore-proposes-another-computers-paths-when-the-two-share-a-hostname) | A restore proposes another computer's paths when the two share a hostname | Not fixed |
-| [B272](#b272--after-a-folder-is-removed-and-added-again-its-file-events-go-to-the-old-id) | After a folder is removed and added again, its file events go to the old id | Not fixed |
 | [B271](#b271--one-failed-listing-makes-a-large-first-sync-start-over-from-nothing) | One failed listing makes a large first sync start over from nothing | Not fixed |
 | [B270](#b270--one-file-left-in-a-mirror-folder-clears-the-wipe-guard-for-all-the-others) | One file left in a mirror folder clears the wipe guard for all the others | Not fixed |
 | [B269](#b269--a-restore-that-skipped-every-folder-reports-success) | A restore that skipped every folder reports success | Not fixed |
@@ -79,10 +78,6 @@ original wording.
 | [B265](#b265--address-keys-whose-token-is-for-an-inactive-account-key-never-open) | Address keys whose token is for an inactive account key never open | Needs an SDK change; not reproduced |
 | [B264](#b264--a-mount-that-fails-after-starting-its-workers-leaves-them-running) | A mount that fails after starting its workers leaves them running | Not fixed |
 | [B263](#b263--a-mount-that-can-never-succeed-retries-every-5-s-and-tells-only-the-journal) | A mount that can never succeed retries every 5 s and tells only the journal | Not fixed |
-| [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 |
-| [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 |
 | [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 +90,8 @@ original wording.
 
 | Entry | Title | What is left |
 |---|---|---|
+| [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 | The first pass still waits for the watch; a failed watch is not shown in the app |
+| [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 |
@@ -116,6 +113,9 @@ a live account since.
 | Entry | Title |
 |---|---|
 | [B261](#b261--a-folder-that-has-never-synced-reads-up-to-date) | A folder that has never synced reads "up to date" |
+| [B272](#b272--after-a-folder-is-removed-and-added-again-its-file-events-go-to-the-old-id) | After a folder is removed and added again, its file events go to the old id |
+| [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 |
+| [B258](#b258--journalctl--p-warning-shows-nothing-whatever-the-daemon-logged) | `journalctl -p warning` shows nothing, whatever the daemon logged |
 | [B256](#b256--pdfs-copy-right-after-a-save-copies-the-old-version) | `pdfs copy` right after a save copies the old version |
 | [B255](#b255--a-synced-folder-whose-files-are-all-gone-here-is-stuck-for-good) | A synced folder whose files are all gone here is stuck for good |
 | [B254](#b254--an-edit-in-the-same-second-as-a-new-upload-never-arrives) | An edit in the same second as a new upload never arrives |
@@ -395,7 +395,7 @@ written before it existed.
 
 ### B272 — After a folder is removed and added again, its file events go to the old id
 
-**Status:** Open
+**Status:** Fixed, not verified live
 **Found:** 2026-10-31, code audit after issue #53.
 **Where:** `rewatch` in `crates/pdfs-fuse/src/sync/engine.rs`
 
@@ -407,8 +407,10 @@ two-minute poll, and the same goes for every later edit until the daemon restart
 path is still wanted, so the old entry stays with the removed folder's id. Its events become
 `Reconcile(old id)`, `sync_folder_get` finds no row, and the request is dropped.
 
-**Fix.** Not done. Compare `(path, id)` pairs, and update the id of a path that is still
+**Fix.** `keep_wanted` keeps a watch whose path is still wanted and gives it the id the
-watched.
+folder list has now.
+
+**Test:** `a_re_added_folder_keeps_its_watch_under_the_new_id` in `pdfs-fuse`.
 
 ---
 
@@ -613,7 +615,8 @@ show it (a state file, and `STATUS=` to systemd).
 
 ### B262 — Watching a large synced folder holds up the sync engine, and a failed watch is never retried
 
-**Status:** Open
+**Status:** Partly fixed. A watch is still set up before the first pass, and a failed watch
+is not shown in the folder's state.
 **Found:** 2026-10-31, code audit after issue #53.
 **Where:** `rewatch` and `engine_loop` in `crates/pdfs-fuse/src/sync/engine.rs`
 
@@ -631,8 +634,16 @@ line, the folder is never watched, and later changes are found only by the two-m
   and they use up watches the next folder needs.
 - Errors the watcher reports later are dropped in its callback.
 
-**Fix.** Not done. Reconcile first, unwatch after a partial failure and retry with a back-off,
+**Fix.**
-turn off symlink following, and show a watch failure in the folder's state.
+- A failed watch is unwatched, which removes the watches it had added under the folder, and
+  is tried again every 10 minutes. The warning names `fs.inotify.max_user_watches`.
+- The watcher no longer follows symlinks.
+- Errors the watcher reports are logged.
+- Not done: the watch still runs before the folder's first pass. Watching first is what keeps
+  a change made during the pass from being missed, so the order stays; the cost is one walk
+  of a new folder's tree. A failed watch is not shown in the folder's state yet.
+
+**Test:** none for the watch failure itself; it needs the inotify limit reached.
 
 ---
 
@@ -667,7 +678,7 @@ expects the new label.
 
 ### B260 — The poll pushes a new folder's first sync behind every other folder
 
-**Status:** Open
+**Status:** Fixed, not verified live
 **Found:** 2026-10-31, code audit after issue #53.
 **Where:** `engine_loop` in `crates/pdfs-fuse/src/sync/engine.rs`
 
@@ -680,14 +691,18 @@ running. When a burst holds both a `ReconcileAll` and a folder's `Reconcile(id)`
 only `reconcile_all`, which goes in id order. A new folder has the highest id, so it always
 goes last.
 
-**Fix.** Not done. Run the folders asked for by id before the rest, and don't queue a poll
+**Fix.** `run_burst` runs the folders a burst names by id first, lowest id first, and then
-while a pass runs.
+every other folder when the burst also holds a `ReconcileAll`. A named folder is read again
+before its pass. A pause that lands between passes now stops the rest of the burst.
+
+**Test:** `named_folders_run_before_the_rest_of_a_burst` in `pdfs-fuse`.
 
 ---
 
 ### 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).
@@ -711,15 +726,30 @@ 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
+**Fix.**
-the supervisor restart a finished engine thread. Log passes at info, and report the folder,
+- Each pass runs under `catch_unwind` (`EngineHealth::contain` in
-the pass age and the queue in `Diagnostics`.
+  `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.
 
 ---
 
 ### B258 — `journalctl -p warning` shows nothing, whatever the daemon logged
 
-**Status:** Open
+**Status:** Fixed, not verified live
 **Found:** 2026-10-31, issue #53: the reporter ran `journalctl --user -u proton-drive.service -b
 -p warning`, as `docs/TROUBLESHOOTING.md` tells them to, and got `-- No entries --`.
 **Where:** `main` in `crates/pdfs-cli/src/main.rs`; `packaging/proton-drive.service`
@@ -732,8 +762,15 @@ stores every line of a service's output at the unit's default priority, info, un
 starts with a `<N>` prefix. So every `WARN` and `ERROR` line is stored as info, and the
 troubleshooting guide's filter hides all of them.
 
-**Fix.** Not done. When `JOURNAL_STREAM` says the output goes to journald, start each line with
+**Fix.** When standard output is the journal stream (`JOURNAL_STREAM` names the same
-the `<N>` prefix for its level.
+device and inode as fd 1), each line starts with the `<N>` prefix for its level, and ANSI colours
+are off. A terminal that merely inherited `JOURNAL_STREAM` keeps the plain format
+(`crates/pdfs-cli/src/journal.rs`).
+
+**Test:** `each_line_opens_with_the_priority_of_its_level` and
+`journal_stream_must_name_the_same_file` in `pdfs-cli`. On 2026-10-31, `pdfs diagnose` run with
+`RUST_LOG=debug` under `systemd-run --user -p StandardOutput=journal` stored its debug lines at
+`PRIORITY=7` and its report at 6.
 
 ---