feat(core,fuse,gui): push daemon events, polish Photos, adopt newer widgets

The app and the tray stop polling. pdfs-core adds Request::Subscribe: the
connection stays open and carries Event lines (status, transfers, and
Changed with a Topic), with a heartbeat while quiet. In pdfs-fuse, an
EventHub gives each subscriber a mailbox, code that changes something
publishes its topic, and a sampler publishes the status, transfers and sync
progress when they differ. The app follows the feed on one thread, paints
status and transfers as they arrive, and reloads the page on screen when its
topic changes, including after changes made with the pdfs CLI. A hidden
window keeps the changes and catches up when focused. Front-ends fall back
to polling a daemon that predates subscriptions. Refresh buttons leave the
page headers; F5 and Ctrl+R still reload from the server.

Photos gets a scrubber overlaid on the right edge of the timeline in place
of the slider and scrollbar. It shows the years, appears when the pointer
comes near or the timeline scrolls, names the month under the pointer and
jumps to it on drag. Upload becomes a split button whose menu holds the
Google Photos import. Tiles are pooled per row and rebound instead of
rebuilt, hover scales the thumbnail to 1.02 without a shadow, and tiles
lose their tooltip.

On libadwaita 1.8 or newer the app uses AdwSpinner, AdwToggleGroup,
AdwButtonRow and AdwShortcutsDialog, and keeps look-alike stand-ins on
older releases (src/compat.rs, widgets/compat.rs). It still builds against
libadwaita 1.5 and picks at run time, creating the newer widgets by class
name through GtkBuilder, so every package gets them. PDFS_STAND_IN_WIDGETS
forces the stand-ins.

The Sync and share-link icons draw in full again: GTK drops group
transforms from symbolic icons, so their coordinates are baked into the
paths. All catalogs are updated and complete.

crates/pdfs-fuse/src/albums.rs

@@ -15,7 +15,7 @@
 use std::collections::{HashMap, HashSet};
 use std::sync::atomic::Ordering;
 
-use pdfs_core::control::{AlbumInfo, PhotoItem};
+use pdfs_core::control::{AlbumInfo, PhotoItem, Topic};
 use pdfs_core::db::StoredAlbum;
 use pdfs_core::{CoreError, CoreResult};
 
@@ -137,10 +137,12 @@ impl Core {
         }
         let core = self.clone();
         self.rt.spawn(async move {
-            if let Err(e) = core.refresh_albums().await {
-                warn!(error = %e, "background album refresh failed");
-            }
+            let refreshed = core.refresh_albums().await;
             core.albums_refreshing.store(false, Ordering::SeqCst);
+            match refreshed {
+                Ok(_) => core.events.publish(&[Topic::Photos]),
+                Err(e) => warn!(error = %e, "background album refresh failed"),
+            }
         });
     }
 

crates/pdfs-fuse/src/background.rs

@@ -422,6 +422,10 @@ pub(super) async fn run_event_sync(
             cursor = Some(applied);
         }
         flush_dirty_parents(&core, &dirty);
+        // Another device changed the tree: whatever a front-end lists may have
+        // moved, including what is shared and what is in the trash.
+        core.events
+            .publish(&[Topic::Files, Topic::Trash, Topic::Shares, Topic::Conflicts]);
     }
 }
 
@@ -479,6 +483,7 @@ pub(super) async fn run_photos_event_sync(core: Core) {
             continue;
         };
         apply_photo_events(&core, &events);
+        core.events.publish(&[Topic::Photos]);
         // One cursor write per batch, after the batch is applied: a crash in
         // between replays events whose effects are idempotent.
         if let Err(error) = core.db.set_state_str(PHOTOS_EVENT_CURSOR, head.as_str()) {

crates/pdfs-fuse/src/control.rs

@@ -17,8 +17,8 @@ use std::time::{Duration, Instant};
 
 use pdfs_core::config::AppDirs;
 use pdfs_core::control::{
-    ActivityKind, PhotoMonth, RefreshScope, Request as CtlRequest, Response as CtlResponse,
-    TransferDirection, TrashFailure,
+    ActivityKind, Event, HEARTBEAT, PhotoMonth, RefreshScope, Request as CtlRequest,
+    Response as CtlResponse, Topic, TransferDirection, TrashFailure,
 };
 use pdfs_core::{CoreError, CoreResult};
 use proton_drive_rs::proton_sdk::ids::NodeUid;
@@ -286,6 +286,155 @@ impl Drop for RefreshGuard {
     }
 }
 
+/// The reply to [`CtlRequest::Status`], which subscriptions are also sent
+/// whenever it changes.
+pub(crate) fn status_response(core: &Core, username: &str, mountpoint: &Path) -> CtlResponse {
+    let pins = core.cache.list_pins();
+    let queued = core.db.pending_op_counts().unwrap_or_default();
+    let staging = core.cache.staging_usage();
+    CtlResponse::Status {
+        parked_uploads: queued.parked.max(0) as u64,
+        failing_ops: queued.failing.max(0) as u64,
+        failing_error: queued.last_error.clone(),
+        staged_bytes: staging.bytes,
+        staged_oldest_secs: staging.oldest_secs,
+        paused: core.sync_pause().is_some(),
+        paused_until: core.sync_pause().flatten(),
+        username: username.to_string(),
+        mountpoint: mountpoint.display().to_string(),
+        pinned: pins.len(),
+        used: core.cache.usage(),
+        budget: core.cache.budget(),
+        pins,
+        online: core.online.load(Ordering::Relaxed),
+        pending_uploads: queued.uploads.max(0) as u64,
+        pending_changes: queued.changes.max(0) as u64,
+    }
+}
+
+/// What a front-end shows that `request` changes when it succeeds. Requests
+/// from the CLI publish the same way as the app's, so an open window follows
+/// a change made in a terminal.
+fn touched_topics(request: &CtlRequest) -> &'static [Topic] {
+    use CtlRequest as R;
+    match request {
+        R::Pin { .. } | R::Unpin { .. } => &[Topic::Files],
+        R::CreateAlbum { .. }
+        | R::RenameAlbum { .. }
+        | R::DeleteAlbum { .. }
+        | R::AddToAlbum { .. }
+        | R::RemoveFromAlbum { .. }
+        | R::SetPhotoFavorite { .. }
+        | R::UploadPhoto { .. } => &[Topic::Photos],
+        R::TrashNodes { .. } => &[Topic::Files, Topic::Photos, Topic::Trash, Topic::Conflicts],
+        R::Rename { .. } | R::Move { .. } | R::CreateFolder { .. } | R::UploadPaths { .. } => {
+            &[Topic::Files, Topic::Conflicts]
+        }
+        R::Delete { .. } => &[Topic::Files, Topic::Trash, Topic::Conflicts],
+        R::Restore { .. } | R::DeleteForever { .. } | R::EmptyTrash => {
+            &[Topic::Files, Topic::Photos, Topic::Trash, Topic::Conflicts]
+        }
+        R::Refresh { scope } => match scope {
+            RefreshScope::Dir { .. } => &[Topic::Files],
+            RefreshScope::Trash => &[Topic::Trash],
+            // A photos refresh answers once it has started; the timeline
+            // publishes again when it lands.
+            RefreshScope::Photos { .. } => &[],
+        },
+        R::RenameDevice { .. } | R::DeleteDevice { .. } => &[Topic::Devices],
+        R::AdoptDevice { .. } => &[Topic::Devices, Topic::Locations],
+        R::AddSyncFolder { .. }
+        | R::RemoveSyncFolder { .. }
+        | R::SetSyncFolderMode { .. }
+        | R::SetSyncFolderPaused { .. }
+        | R::RestoreSyncFolders { .. }
+        | R::RestoreDeviceFolders { .. } => &[Topic::Locations, Topic::Devices],
+        R::RetryPendingOp { .. } => &[Topic::Queue],
+        R::ResolveConflict { .. } => &[Topic::Files, Topic::Trash, Topic::Conflicts],
+        R::ShareNode { .. }
+        | R::ShareNodeByUid { .. }
+        | R::UpdateShareRole { .. }
+        | R::UpdateShareRoleByUid { .. }
+        | R::RemoveShareEntry { .. }
+        | R::RemoveShareEntryByUid { .. }
+        | R::CreatePublicLink { .. }
+        | R::CreatePublicLinkByUid { .. }
+        | R::RemovePublicLink { .. }
+        | R::RemovePublicLinkByUid { .. }
+        | R::StopSharing { .. }
+        | R::StopSharingByUid { .. } => &[Topic::Shares, Topic::Files],
+        R::LeaveShared { .. }
+        | R::AcceptInvitation { .. }
+        | R::RejectInvitation { .. }
+        | R::CreateBookmark { .. }
+        | R::DeleteBookmark { .. } => &[Topic::Shares],
+        R::RestoreRevision { .. }
+        | R::RestoreRevisionByUid { .. }
+        | R::SaveRevisionAs { .. }
+        | R::SaveRevisionAsByUid { .. } => &[Topic::Files],
+        _ => &[],
+    }
+}
+
+/// How long a subscription's writes may block. A client that stops reading
+/// is dropped after this rather than holding its handler forever.
+const SUBSCRIPTION_WRITE_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// Serve a [`CtlRequest::Subscribe`]: send the current status and transfers
+/// and say every topic changed, so the client catches up, then send whatever
+/// the [`EventHub`](super::events::EventHub) collects, with a heartbeat when
+/// nothing does. Ends when the client goes away or the daemon stops.
+fn serve_subscription(core: &Core, username: &str, mountpoint: &Path, mut stream: UnixStream) {
+    let Some(mailbox) = core.events.subscribe() else {
+        warn!("control: refusing a subscription at the limit");
+        if let Ok(mut out) = serde_json::to_vec(&CtlResponse::error(CoreError::remote(
+            "too many subscriptions",
+        ))) {
+            out.push(b'\n');
+            let _ = stream.write_all(&out);
+        }
+        return;
+    };
+    struct Unsubscribe<'a>(&'a Core, std::sync::Arc<super::events::Mailbox>);
+    impl Drop for Unsubscribe<'_> {
+        fn drop(&mut self) {
+            self.0.events.unsubscribe(&self.1);
+        }
+    }
+    let _unsubscribe = Unsubscribe(core, mailbox.clone());
+    if let Err(e) = stream.set_write_timeout(Some(SUBSCRIPTION_WRITE_TIMEOUT)) {
+        warn!(error = %e, "control: could not set subscription timeout");
+        return;
+    }
+    let heartbeat = super::events::event_line(&Event::Heartbeat);
+    let mut lines: Vec<_> = [
+        super::events::event_line(&Event::Status(Box::new(status_response(
+            core, username, mountpoint,
+        )))),
+        super::events::transfers_line(core.transfers.snapshot(), core.jobs_snapshot()),
+        super::events::event_line(&Event::Changed {
+            topics: Topic::ALL.to_vec(),
+        }),
+    ]
+    .into_iter()
+    .flatten()
+    .collect();
+    loop {
+        for line in &lines {
+            if stream.write_all(line.as_bytes()).is_err() {
+                return;
+            }
+        }
+        if core.shutdown.is_stopping() {
+            return;
+        }
+        lines = mailbox.take(HEARTBEAT);
+        if lines.is_empty() {
+            lines.extend(heartbeat.clone());
+        }
+    }
+}
+
 fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: UnixStream) {
     if let Err(e) = stream.set_read_timeout(Some(CONTROL_READ_TIMEOUT)) {
         warn!(error = %e, "control: could not set request timeout");
@@ -310,32 +459,21 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
             return;
         }
     };
+    let request = serde_json::from_str::<CtlRequest>(line.trim());
+    // A subscription lasts as long as its client, so it is not a request in
+    // flight and stays out of the diagnostics' list of them.
+    if let Ok(CtlRequest::Subscribe) = request {
+        serve_subscription(core, username, mountpoint, stream);
+        return;
+    }
+    let touched = request.as_ref().map_or(&[][..], touched_topics);
     // Registered before the request is even parsed, so a handler that wedges
     // inside the daemon shows up by name in `pdfs diagnostics`.
     let _inflight = super::diagnostics::InflightGuard::new(super::diagnostics::request_kind(&line));
-    let response = match serde_json::from_str::<CtlRequest>(line.trim()) {
-        Ok(CtlRequest::Status) => {
-            let pins = core.cache.list_pins();
-            let queued = core.db.pending_op_counts().unwrap_or_default();
-            let staging = core.cache.staging_usage();
-            CtlResponse::Status {
-                parked_uploads: queued.parked.max(0) as u64,
-                failing_ops: queued.failing.max(0) as u64,
-                failing_error: queued.last_error.clone(),
-                staged_bytes: staging.bytes,
-                staged_oldest_secs: staging.oldest_secs,
-                paused: core.sync_pause().is_some(),
-                paused_until: core.sync_pause().flatten(),
-                username: username.to_string(),
-                mountpoint: mountpoint.display().to_string(),
-                pinned: pins.len(),
-                used: core.cache.usage(),
-                budget: core.cache.budget(),
-                pins,
-                online: core.online.load(Ordering::Relaxed),
-                pending_uploads: queued.uploads.max(0) as u64,
-                pending_changes: queued.changes.max(0) as u64,
-            }
+    let response = match request {
+        Ok(CtlRequest::Status) => status_response(core, username, mountpoint),
+        Ok(CtlRequest::Subscribe) => {
+            CtlResponse::error(CoreError::invalid("a subscription is served on its own"))
         }
         Ok(CtlRequest::Pin { path }) => match rel_to_mount(mountpoint, &path) {
             Ok(rel) => match core.pin(&rel) {
@@ -1584,6 +1722,9 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
         // The request did not parse: the caller sent something malformed.
         Err(e) => CtlResponse::error(CoreError::invalid(format!("bad request: {e}"))),
     };
+    if !touched.is_empty() && !matches!(response, CtlResponse::Error { .. }) {
+        core.events.publish(touched);
+    }
     let mut out = match serde_json::to_vec(&response) {
         Ok(v) => v,
         Err(e) => {

crates/pdfs-fuse/src/drain.rs

@@ -1184,6 +1184,10 @@ impl Core {
             format!("{reason}; local changes uploaded as {alt}"),
             false,
         );
+        self.events.publish(&[
+            pdfs_core::control::Topic::Files,
+            pdfs_core::control::Topic::Conflicts,
+        ]);
         info!(%uid, name, alt, "queued write landed as a conflict copy");
         Ok(())
     }

crates/pdfs-fuse/src/events.rs

@@ -0,0 +1,321 @@
+//! Push notifications for front-ends: the daemon's side of
+//! [`Request::Subscribe`](pdfs_core::control::Request::Subscribe).
+//!
+//! Front-ends used to poll. The app asked for the status every two seconds, the
+//! tray every three, and pages re-read themselves on a timer, whether or not
+//! anything had changed. Now each subscription owns a [`Mailbox`] in the
+//! [`EventHub`]. Code that changes something publishes a [`Topic`], and
+//! [`run_event_sampler`] watches the few things that change without passing
+//! through one place (the status, transfers, sync progress) and publishes them
+//! when they differ.
+
+use std::collections::BTreeSet;
+use std::sync::Arc;
+use std::time::Duration;
+
+use parking_lot::{Condvar, Mutex, MutexGuard};
+use pdfs_core::control::{Event, JobItem, Response as CtlResponse, Topic, TransferItem};
+use tracing::warn;
+
+use super::Core;
+
+/// Most subscriptions served at once. Each holds a control handler for as long
+/// as it lasts; the app and the tray need one apiece, and the cap keeps a
+/// misbehaving client from spending the handlers that requests need.
+const MAX_SUBSCRIBERS: usize = 8;
+
+/// How long a subscription waits after a change before it sends, so a burst
+/// (a folder of files trashed at once) goes out as one event.
+const COALESCE: Duration = Duration::from_millis(150);
+
+/// How often [`run_event_sampler`] looks at transfers. Status and sync
+/// progress are looked at every other time.
+const SAMPLE_INTERVAL: Duration = Duration::from_millis(500);
+
+/// Every open subscription's mailbox. Shared by every clone of [`Core`],
+/// on-demand forks included, so a change in any mount reaches every client.
+#[derive(Default)]
+pub(crate) struct EventHub {
+    mailboxes: Mutex<Vec<Arc<Mailbox>>>,
+}
+
+/// What one subscription has yet to send.
+#[derive(Default)]
+pub(crate) struct Mailbox {
+    inbox: Mutex<Inbox>,
+    ready: Condvar,
+}
+
+#[derive(Default)]
+struct Inbox {
+    topics: BTreeSet<Topic>,
+    /// The newest [`Event::Status`] line. A newer one replaces it unsent: a
+    /// client wants the current status, not every status in between.
+    status: Option<Arc<str>>,
+    /// The newest [`Event::Transfers`] line, replaced the same way.
+    transfers: Option<Arc<str>>,
+}
+
+impl Inbox {
+    fn is_empty(&self) -> bool {
+        self.topics.is_empty() && self.status.is_none() && self.transfers.is_none()
+    }
+}
+
+/// `event` as one line of the wire format, newline included.
+pub(crate) fn event_line(event: &Event) -> Option<Arc<str>> {
+    match serde_json::to_string(event) {
+        Ok(mut line) => {
+            line.push('\n');
+            Some(line.into())
+        }
+        Err(e) => {
+            warn!(error = %e, "events: serialize event failed");
+            None
+        }
+    }
+}
+
+impl EventHub {
+    /// Tell every subscriber that `topics` changed.
+    pub(crate) fn publish(&self, topics: &[Topic]) {
+        self.deliver(|inbox| inbox.topics.extend(topics.iter().copied()));
+    }
+
+    /// Hand every subscriber a fresh status.
+    fn publish_status(&self, line: Arc<str>) {
+        self.deliver(|inbox| inbox.status = Some(line.clone()));
+    }
+
+    /// Hand every subscriber a fresh transfers snapshot.
+    fn publish_transfers(&self, line: Arc<str>) {
+        self.deliver(|inbox| inbox.transfers = Some(line.clone()));
+    }
+
+    fn deliver(&self, put: impl Fn(&mut Inbox)) {
+        for mailbox in self.mailboxes.lock().iter() {
+            put(&mut mailbox.inbox.lock());
+            mailbox.ready.notify_one();
+        }
+    }
+
+    /// Open a mailbox, or `None` at [`MAX_SUBSCRIBERS`].
+    pub(crate) fn subscribe(&self) -> Option<Arc<Mailbox>> {
+        let mut mailboxes = self.mailboxes.lock();
+        if mailboxes.len() >= MAX_SUBSCRIBERS {
+            return None;
+        }
+        let mailbox = Arc::new(Mailbox::default());
+        mailboxes.push(mailbox.clone());
+        Some(mailbox)
+    }
+
+    pub(crate) fn unsubscribe(&self, mailbox: &Arc<Mailbox>) {
+        self.mailboxes
+            .lock()
+            .retain(|other| !Arc::ptr_eq(other, mailbox));
+    }
+
+    fn has_subscribers(&self) -> bool {
+        !self.mailboxes.lock().is_empty()
+    }
+}
+
+impl Mailbox {
+    /// Wait up to `timeout` for something to send, then take it as wire lines:
+    /// the status first, then transfers, then the changed topics. Empty when
+    /// nothing arrived in time.
+    pub(crate) fn take(&self, timeout: Duration) -> Vec<Arc<str>> {
+        let mut inbox = self.inbox.lock();
+        if inbox.is_empty() {
+            self.ready.wait_for(&mut inbox, timeout);
+            if inbox.is_empty() {
+                return Vec::new();
+            }
+            // Let the rest of a burst arrive before sending the first of it.
+            MutexGuard::unlocked(&mut inbox, || std::thread::sleep(COALESCE));
+        }
+        let mut lines = Vec::new();
+        lines.extend(inbox.status.take());
+        lines.extend(inbox.transfers.take());
+        if !inbox.topics.is_empty() {
+            let topics = std::mem::take(&mut inbox.topics).into_iter().collect();
+            lines.extend(event_line(&Event::Changed { topics }));
+        }
+        lines
+    }
+}
+
+/// The [`Event::Transfers`] line for a snapshot.
+pub(crate) fn transfers_line(items: Vec<TransferItem>, jobs: Vec<JobItem>) -> Option<Arc<str>> {
+    event_line(&Event::Transfers { items, jobs })
+}
+
+/// Watch what changes without a single place to publish it from, and publish it
+/// when it differs from the last look: the status (queue counts, cache use,
+/// online, pause), the transfers snapshot, and the synced folders with their
+/// live progress. Looks only while someone is subscribed.
+pub(crate) fn run_event_sampler(core: Core, username: String, mountpoint: std::path::PathBuf) {
+    let mut transfers: Option<Arc<str>> = None;
+    let mut status: Option<String> = None;
+    let mut queue: Option<String> = None;
+    let mut locations: Option<String> = None;
+    let mut tick = 0u32;
+    while core.shutdown.sleep(SAMPLE_INTERVAL) {
+        if !core.events.has_subscribers() {
+            // A new subscription starts from a snapshot of its own, so there is
+            // nothing to compare against until then.
+            transfers = None;
+            status = None;
+            queue = None;
+            locations = None;
+            continue;
+        }
+        let line = transfers_line(core.transfers.snapshot(), core.jobs_snapshot());
+        if line.is_some() && line != transfers {
+            transfers = line.clone();
+            if let Some(line) = line {
+                core.events.publish_transfers(line);
+            }
+        }
+
+        tick = tick.wrapping_add(1);
+        if !tick.is_multiple_of(2) {
+            continue;
+        }
+        let now = super::control::status_response(&core, &username, &mountpoint);
+        let (key, queue_key) = status_keys(&now);
+        if status.as_ref().is_some_and(|last| *last != key)
+            && let Some(line) = event_line(&Event::Status(Box::new(now)))
+        {
+            core.events.publish_status(line);
+        }
+        status = Some(key);
+        if queue.as_ref().is_some_and(|last| *last != queue_key) {
+            core.events.publish(&[Topic::Queue]);
+        }
+        queue = Some(queue_key);
+
+        if let Ok(now) = core.list_locations() {
+            let key = serde_json::to_string(&now).unwrap_or_default();
+            if locations.as_ref().is_some_and(|last| *last != key) {
+                core.events.publish(&[Topic::Locations]);
+            }
+            locations = Some(key);
+        }
+    }
+}
+
+/// What makes two statuses different enough to send, and what makes their
+/// queues different. The age of the oldest staged write is left out of both:
+/// it grows every second while anything is staged, which is not news.
+fn status_keys(status: &CtlResponse) -> (String, String) {
+    let CtlResponse::Status {
+        parked_uploads,
+        failing_ops,
+        failing_error,
+        staged_bytes,
+        paused,
+        paused_until,
+        username,
+        mountpoint,
+        pinned,
+        used,
+        budget,
+        online,
+        pending_uploads,
+        pending_changes,
+        ..
+    } = status
+    else {
+        return (String::new(), String::new());
+    };
+    let queue = format!(
+        "{pending_uploads}/{pending_changes}/{parked_uploads}/{failing_ops}/{failing_error:?}"
+    );
+    let key = format!(
+        "{queue}/{staged_bytes}/{paused}/{paused_until:?}/{username}/{mountpoint}/{pinned}/\
+         {used}/{budget}/{online}"
+    );
+    (key, queue)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn changed(line: &str) -> Vec<Topic> {
+        match serde_json::from_str::<Event>(line.trim()) {
+            Ok(Event::Changed { topics }) => topics,
+            other => panic!("not a change event: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn a_burst_of_changes_arrives_as_one_event() {
+        let hub = EventHub::default();
+        let mailbox = hub.subscribe().expect("room for a subscriber");
+        hub.publish(&[Topic::Trash, Topic::Files]);
+        hub.publish(&[Topic::Files]);
+        let lines = mailbox.take(Duration::from_secs(1));
+        assert_eq!(lines.len(), 1);
+        assert_eq!(changed(&lines[0]), vec![Topic::Files, Topic::Trash]);
+        assert!(mailbox.take(Duration::from_millis(10)).is_empty());
+    }
+
+    #[test]
+    fn a_newer_status_replaces_one_not_yet_sent() {
+        let hub = EventHub::default();
+        let mailbox = hub.subscribe().expect("room for a subscriber");
+        hub.publish_status("first\n".into());
+        hub.publish_status("second\n".into());
+        assert_eq!(
+            mailbox.take(Duration::from_secs(1)),
+            vec![Arc::<str>::from("second\n")]
+        );
+    }
+
+    #[test]
+    fn a_closed_subscription_gets_nothing_more() {
+        let hub = EventHub::default();
+        let mailbox = hub.subscribe().expect("room for a subscriber");
+        hub.unsubscribe(&mailbox);
+        hub.publish(&[Topic::Files]);
+        assert!(mailbox.take(Duration::from_millis(10)).is_empty());
+        assert!(!hub.has_subscribers());
+    }
+
+    #[test]
+    fn subscriptions_are_capped() {
+        let hub = EventHub::default();
+        let open: Vec<_> = (0..MAX_SUBSCRIBERS)
+            .map(|_| hub.subscribe().expect("room for a subscriber"))
+            .collect();
+        assert!(hub.subscribe().is_none());
+        hub.unsubscribe(&open[0]);
+        assert!(hub.subscribe().is_some());
+    }
+
+    #[test]
+    fn the_age_of_a_staged_write_is_not_a_change() {
+        let status = |staged_oldest_secs| CtlResponse::Status {
+            parked_uploads: 0,
+            failing_ops: 0,
+            failing_error: None,
+            staged_bytes: 10,
+            staged_oldest_secs,
+            paused: false,
+            paused_until: None,
+            username: "u".into(),
+            mountpoint: "/m".into(),
+            pinned: 0,
+            used: 0,
+            budget: 0,
+            pins: Vec::new(),
+            online: true,
+            pending_uploads: 1,
+            pending_changes: 0,
+        };
+        assert_eq!(status_keys(&status(5)), status_keys(&status(6)));
+    }
+}

crates/pdfs-fuse/src/lib.rs

@@ -60,7 +60,7 @@ use pdfs_core::config::{AppDirs, SweepMode};
 use pdfs_core::control::{
     ActivityEntry, ActivityKind, DirEntry, ErrorKind, LocalHit, PhotoKind, PublicLinkInfo,
     SearchFilters, SearchHit, SearchSource, SyncFolderInfo, SyncPhase, SyncProgress,
-    ThumbnailBuildStatus, TransferDirection,
+    ThumbnailBuildStatus, Topic, TransferDirection,
 };
 use pdfs_core::db::{
     Db, LOCAL_VOLUME, OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PARK_UNTIL, PendingOp,
@@ -86,6 +86,7 @@ use control::run_control_socket;
 mod devices;
 mod diagnostics;
 mod drain;
+mod events;
 mod filesystem;
 pub use filesystem::ProtonFs;
 mod mount;
@@ -665,6 +666,9 @@ struct Core {
     /// In-flight upload/download progress, served to `GetQueueStatus`. Shared
     /// across the FUSE session and the control-socket task.
     transfers: Arc<TransferRegistry>,
+    /// Open [`Request::Subscribe`](pdfs_core::control::Request::Subscribe)
+    /// connections, which every change a front-end shows is published to.
+    events: Arc<events::EventHub>,
     /// True while the background scanner is rebuilding the local-file index, so
     /// `SearchLocal` can tell a front-end "still indexing" apart from "no match".
     indexing: Arc<AtomicBool>,
@@ -4533,10 +4537,12 @@ impl Core {
         }
         let core = self.clone();
         self.rt.spawn(async move {
-            if let Err(e) = core.refresh_trash().await {
-                warn!(error = %e, "background trash refresh failed");
-            }
+            let refreshed = core.refresh_trash().await;
             core.trash_refreshing.store(false, Ordering::SeqCst);
+            match refreshed {
+                Ok(_) => core.events.publish(&[Topic::Trash]),
+                Err(e) => warn!(error = %e, "background trash refresh failed"),
+            }
             // After the flag, so a waiter that wakes on this sees it cleared.
             core.trash_progress.notify_waiters();
         });
@@ -4827,6 +4833,14 @@ impl Core {
         };
         if let Err(e) = self.db.activity_add(&entry) {
             warn!(error = ?e, "could not record activity");
+            return;
+        }
+        // A sync pass is how a conflict copy reaches the listing, and the sweep
+        // reports the ones it finds as conflicts.
+        if matches!(entry.kind, ActivityKind::Conflict | ActivityKind::Sync) {
+            self.events.publish(&[Topic::Activity, Topic::Conflicts]);
+        } else {
+            self.events.publish(&[Topic::Activity]);
         }
     }
 

crates/pdfs-fuse/src/mount.rs

@@ -416,6 +416,7 @@ pub fn mount(
         notifier: Arc::new(OnceLock::new()),
         session_live: Arc::new(AtomicBool::new(false)),
         transfers: TransferRegistry::new(),
+        events: Arc::new(Default::default()),
         indexing: Arc::new(AtomicBool::new(false)),
         sync_progress: Arc::new(Mutex::new(HashMap::new())),
         sync_tx,
@@ -568,6 +569,22 @@ pub fn mount(
             }
         }
     }
+    // Publishes what changes without one place to publish it from (status,
+    // transfers, sync progress) to the front-ends subscribed on that socket.
+    {
+        let sampler_core = core.clone();
+        let username = username.clone();
+        let mountpoint = mountpoint.to_path_buf();
+        match std::thread::Builder::new()
+            .name("pdfs-events".into())
+            .spawn(move || events::run_event_sampler(sampler_core, username, mountpoint))
+        {
+            Ok(handle) => workers.push(handle),
+            // Not fatal: front-ends still get every published change, only
+            // not the sampled ones.
+            Err(error) => warn!(error = %error, "could not start the event sampler"),
+        }
+    }
     // The supervisor watches the pool and answers the systemd watchdog. Started
     // after the listener, because its liveness probe is a round trip over that
     // socket.

crates/pdfs-fuse/src/photos.rs

@@ -19,7 +19,7 @@ use std::time::{Duration, Instant};
 use base64::Engine as _;
 use pdfs_core::control::{
     FileThumbRequest, MappingProgress, PhotoItem, PhotoKind, PhotoLocation, PhotoThumb, PlaceInfo,
-    ThumbnailBuildStatus, is_raw_image_name, is_thumbnail_image_name,
+    ThumbnailBuildStatus, Topic, is_raw_image_name, is_thumbnail_image_name,
 };
 use pdfs_core::db::{self, StoredPhoto};
 use pdfs_core::{CoreError, CoreResult};
@@ -2041,10 +2041,14 @@ impl Core {
         }
         let core = self.clone();
         self.rt.spawn(async move {
-            if let Err(e) = core.refresh_timeline().await {
-                warn!(error = %e, "background timeline refresh failed");
-            }
+            let refreshed = core.refresh_timeline().await;
             core.timeline_refreshing.store(false, Ordering::SeqCst);
+            // Only a refresh that landed: a front-end re-reading after a failed
+            // one would find the timeline still stale and start another.
+            match refreshed {
+                Ok(_) => core.events.publish(&[Topic::Photos]),
+                Err(e) => warn!(error = %e, "background timeline refresh failed"),
+            }
         });
     }