persistent entity cache, streaming batche

crates/pdfs-fuse/Cargo.toml

@@ -10,6 +10,7 @@ description = "FUSE on-demand filesystem for Proton Drive"
 [dependencies]
 pdfs-core.workspace = true
 proton-drive-rs.workspace = true
+futures.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 fuser.workspace = true

crates/pdfs-fuse/src/background.rs

@@ -358,6 +358,14 @@ pub(super) async fn run_event_sync(
         // started offline (offline.md Phase 1) — so retry rather than giving up,
         // which used to disable live sync for the life of the daemon.
         Ok(None) => {
+            // Nothing says what changed before this cursor, so a persisted SDK
+            // entity cache from an earlier run cannot be trusted: drop it rather
+            // than serve metadata no event will ever invalidate.
+            if let Some(cache) = pdfs_core::sdkcache::SdkCache::opened()
+                && let Err(e) = cache.clear_now()
+            {
+                warn!(error = %e, "clearing the persisted SDK entity cache failed");
+            }
             let mut delay = ONLINE_PROBE_MIN;
             loop {
                 match client.enumerate_events(&scope, None).await {

crates/pdfs-fuse/src/control.rs

@@ -203,7 +203,8 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
             limit,
             kind,
             range,
-        }) => match core.photos_timeline(offset, limit, kind, range) {
+            favorites,
+        }) => match core.photos_timeline(offset, limit, kind, range, favorites) {
             Ok(Some(items)) => CtlResponse::Photos {
                 available: true,
                 items,
@@ -216,6 +217,19 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
             },
             Err(e) => CtlResponse::error(e),
         },
+        Ok(CtlRequest::SetPhotoFavorite { uid, favorite }) => match parse_uid(&uid) {
+            Some(node_uid) => match core.set_photo_favorite(&node_uid, favorite) {
+                Ok(()) => CtlResponse::Ok {
+                    message: if favorite {
+                        "added to favourites".to_string()
+                    } else {
+                        "removed from favourites".to_string()
+                    },
+                },
+                Err(e) => CtlResponse::error(e),
+            },
+            None => CtlResponse::error(CoreError::invalid(format!("invalid uid: {uid}"))),
+        },
         Ok(CtlRequest::PhotoAlbums) => match core.albums() {
             Ok(Some(items)) => CtlResponse::Albums {
                 available: true,
@@ -863,6 +877,78 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
                 Err(e) => CtlResponse::error(e),
             }
         }
+        Ok(CtlRequest::ListRevisions { path }) => match rel_to_mount(mountpoint, &path) {
+            Ok(rel) => match core.list_revisions(&rel) {
+                Ok(items) => CtlResponse::Revisions { items },
+                Err(e) => CtlResponse::error(e),
+            },
+            Err(e) => CtlResponse::error(e),
+        },
+        Ok(CtlRequest::ListRevisionsByUid { uid }) => match core.list_revisions_by_uid(&uid) {
+            Ok(items) => CtlResponse::Revisions { items },
+            Err(e) => CtlResponse::error(e),
+        },
+        Ok(CtlRequest::RestoreRevision { path, revision_id }) => {
+            match rel_to_mount(mountpoint, &path) {
+                Ok(rel) => match core.restore_revision(&rel, &revision_id) {
+                    Ok(()) => CtlResponse::Ok {
+                        message: format!("restoring version {revision_id} of {path}"),
+                    },
+                    Err(e) => CtlResponse::error(e),
+                },
+                Err(e) => CtlResponse::error(e),
+            }
+        }
+        Ok(CtlRequest::RestoreRevisionByUid { uid, revision_id }) => {
+            match core.restore_revision_by_uid(&uid, &revision_id) {
+                Ok(()) => CtlResponse::Ok {
+                    message: format!("restoring version {revision_id}"),
+                },
+                Err(e) => CtlResponse::error(e),
+            }
+        }
+        Ok(CtlRequest::DeleteRevision { path, revision_id }) => {
+            match rel_to_mount(mountpoint, &path) {
+                Ok(rel) => match core.delete_revision(&rel, &revision_id) {
+                    Ok(()) => CtlResponse::Ok {
+                        message: format!("deleted version {revision_id} of {path}"),
+                    },
+                    Err(e) => CtlResponse::error(e),
+                },
+                Err(e) => CtlResponse::error(e),
+            }
+        }
+        Ok(CtlRequest::DeleteRevisionByUid { uid, revision_id }) => {
+            match core.delete_revision_by_uid(&uid, &revision_id) {
+                Ok(()) => CtlResponse::Ok {
+                    message: format!("deleted version {revision_id}"),
+                },
+                Err(e) => CtlResponse::error(e),
+            }
+        }
+        Ok(CtlRequest::SaveRevisionAs {
+            path,
+            revision_id,
+            dest,
+        }) => match rel_to_mount(mountpoint, &path) {
+            Ok(rel) => match core.save_revision_as(&rel, &revision_id, Path::new(&dest)) {
+                Ok(written) => CtlResponse::Ok {
+                    message: format!("wrote {} to {dest}", human_bytes(written)),
+                },
+                Err(e) => CtlResponse::error(e),
+            },
+            Err(e) => CtlResponse::error(e),
+        },
+        Ok(CtlRequest::SaveRevisionAsByUid {
+            uid,
+            revision_id,
+            dest,
+        }) => match core.save_revision_as_by_uid(&uid, &revision_id, Path::new(&dest)) {
+            Ok(written) => CtlResponse::Ok {
+                message: format!("wrote {} to {dest}", human_bytes(written)),
+            },
+            Err(e) => CtlResponse::error(e),
+        },
         Ok(CtlRequest::ListSharedWithMe) => match core.list_shared_with_me() {
             Ok(entries) => CtlResponse::Entries { entries },
             Err(e) => CtlResponse::error(e),

crates/pdfs-fuse/src/devices.rs

@@ -12,6 +12,7 @@ use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 use parking_lot::Mutex;
+use pdfs_core::batch;
 use pdfs_core::control::{
     ActivityKind, DeviceInfo, JobItem, MountKind, MountSpec, SyncFolderInfo, SyncPhase,
     SyncProgress,
@@ -557,7 +558,10 @@ impl Core {
         let _ = self.sync_tx.send(sync::SyncMsg::Rewatch);
         if delete_remote
             && let Some(uid) = parse_uid(&folder.remote_uid)
-            && let Err(e) = self.rt.block_on(self.client.trash_nodes(&[uid]))
+            && let Err(e) = self
+                .rt
+                .block_on(self.client.trash_nodes(&[uid]))
+                .and_then(batch::into_unit)
         {
             warn!(id, error = %e, "delete remote device folder failed");
         }

crates/pdfs-fuse/src/drain.rs

@@ -25,6 +25,7 @@ use std::path::{Path, PathBuf};
 use std::sync::atomic::Ordering;
 use std::time::Duration;
 
+use pdfs_core::batch;
 use pdfs_core::cache::{Baseline, StagedWrite};
 use pdfs_core::control::{ActivityKind, TransferDirection};
 use pdfs_core::db::{OP_CREATE, OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PendingOp, RenameMeta};
@@ -340,6 +341,7 @@ impl Core {
         match self
             .rt
             .block_on(self.client.trash_nodes(std::slice::from_ref(&uid)))
+            .and_then(batch::into_unit)
         {
             Ok(()) => {}
             Err(e) if is_gone(&e) => {

crates/pdfs-fuse/src/filesystem.rs

@@ -1856,6 +1856,7 @@ impl ProtonFs {
             .core
             .rt
             .block_on(self.core.client.trash_nodes(std::slice::from_ref(&uid)))
+            .and_then(batch::into_unit)
         {
             error!(%uid, error = %e, "trash failed");
             self.core

crates/pdfs-fuse/src/lib.rs

@@ -53,6 +53,8 @@ use fuser::{
     ReplyDirectory, ReplyEmpty, ReplyEntry, ReplyIoctl, ReplyLseek, ReplyOpen, ReplyWrite, Request,
     Session, TimeOrNow, WriteFlags,
 };
+use futures::StreamExt as _;
+use pdfs_core::batch;
 use pdfs_core::cache::{BLOCK_SIZE, Baseline, ContentCache, StagedWrite};
 use pdfs_core::config::{AppDirs, SweepMode};
 use pdfs_core::control::{
@@ -88,6 +90,7 @@ mod mount;
 mod photos;
 mod profile;
 mod reads;
+mod revisions;
 mod sharing;
 mod state;
 mod sweep;
@@ -3287,6 +3290,7 @@ impl Core {
         .map_err(|error| self.errno_error(error, "trash access"))?;
         self.rt
             .block_on(self.client.trash_nodes(std::slice::from_ref(&uid)))
+            .and_then(batch::into_unit)
             .map_err(|e| CoreError::from_api(&e, "trash"))?;
         let name = self
             .state
@@ -3325,6 +3329,7 @@ impl Core {
         if let Err(e) = self
             .rt
             .block_on(self.client.trash_nodes(std::slice::from_ref(uid)))
+            .and_then(batch::into_unit)
         {
             error!(%uid, name, error = %e, "trashing the node a rename replaces failed");
             self.log_activity(ActivityKind::Trash, name, e.to_string(), false);
@@ -3372,6 +3377,7 @@ impl Core {
         match self
             .rt
             .block_on(self.client.restore_nodes(std::slice::from_ref(uid)))
+            .and_then(batch::into_unit)
         {
             Ok(()) => {
                 self.invalidate_trash();
@@ -3586,14 +3592,37 @@ impl Core {
             .into_iter()
             .filter_map(|n| n.parent_uid)
             .collect();
+        // Streamed per-node outcomes: a uid the server refuses (already restored,
+        // gone) leaves the rest of the batch restored, and each node is unhidden
+        // as *its* batch lands rather than after the last one — a daemon killed
+        // half way through a large restore leaves local state describing what the
+        // server actually did. Only a batch that restored nothing is an error the
+        // caller can act on.
+        let mut restored: Vec<NodeUid> = Vec::with_capacity(parsed.len());
+        let mut first_error: Option<ProtonError> = None;
         self.rt
-            .block_on(self.client.restore_nodes(&parsed))
+            .block_on(async {
+                let mut outcomes = std::pin::pin!(self.client.restore_nodes_streaming(&parsed));
+                while let Some(item) = outcomes.next().await {
+                    let (uid, outcome) = item?;
+                    match outcome {
+                        Ok(()) => {
+                            self.hidden.lock().remove(&uid);
+                            restored.push(uid);
+                        }
+                        Err(e) => {
+                            warn!(%uid, error = %e, "restore failed for a node");
+                            first_error.get_or_insert(e);
+                        }
+                    }
+                }
+                Ok::<(), ProtonError>(())
+            })
             .map_err(|e| CoreError::from_api(&e, "restore"))?;
+        if restored.is_empty()
+            && let Some(error) = first_error
         {
-            let mut hidden = self.hidden.lock();
-            for uid in &parsed {
-                hidden.remove(uid);
-            }
+            return Err(CoreError::from_api(&error, "restore"));
         }
         // Keyed by uid, so it applies to every mount: a restored node reappears
         // in whichever inode spaces show its parent, not only the primary one.
@@ -3605,19 +3634,45 @@ impl Core {
             }
         });
         self.invalidate_trash();
-        Ok(parsed.len())
+        Ok(restored.len())
     }
 
     /// Permanently delete trashed nodes. Irreversible on the server; locally it
     /// drops any metadata and cached content the node still owns.
     fn delete_forever(&self, uids: &[String]) -> CoreResult<usize> {
         let parsed = Self::parse_uids(uids)?;
+        // Streamed, and each node's local state is dropped as its batch lands.
+        // A permanent delete is irreversible, so the useful property is that a
+        // daemon interrupted mid-batch has already forgotten exactly the nodes
+        // the server destroyed — never more, never fewer.
+        let mut deleted = 0usize;
+        let mut first_error: Option<ProtonError> = None;
         self.rt
-            .block_on(self.client.delete_nodes(&parsed))
+            .block_on(async {
+                let mut outcomes = std::pin::pin!(self.client.delete_nodes_streaming(&parsed));
+                while let Some(item) = outcomes.next().await {
+                    let (uid, outcome) = item?;
+                    match outcome {
+                        Ok(()) => {
+                            self.drop_local(std::slice::from_ref(&uid));
+                            deleted += 1;
+                        }
+                        Err(e) => {
+                            warn!(%uid, error = %e, "permanent delete failed for a node");
+                            first_error.get_or_insert(e);
+                        }
+                    }
+                }
+                Ok::<(), ProtonError>(())
+            })
             .map_err(|e| CoreError::from_api(&e, "delete"))?;
-        self.drop_local(&parsed);
+        if deleted == 0
+            && let Some(error) = first_error
+        {
+            return Err(CoreError::from_api(&error, "delete"));
+        }
         self.invalidate_trash();
-        Ok(parsed.len())
+        Ok(deleted)
     }
 
     /// Permanently delete everything in the trash. The uids are listed first so

crates/pdfs-fuse/src/photos.rs

@@ -14,7 +14,7 @@ use pdfs_core::{CoreError, CoreResult};
 use std::sync::atomic::Ordering;
 
 use proton_drive_rs::proton_sdk::ids::NodeUid;
-use proton_drive_rs::{NodeKind, ThumbnailType};
+use proton_drive_rs::{NodeKind, PhotoTag, PhotoTagsUpdate, ThumbnailType};
 use tracing::{info, warn};
 
 use super::{
@@ -97,6 +97,7 @@ impl Core {
         limit: usize,
         kind: Option<PhotoKind>,
         range: Option<(i64, i64)>,
+        favorites: bool,
     ) -> CoreResult<Option<Vec<PhotoItem>>> {
         let count = self.db.photos_count().map_err(CoreError::from)?;
         if count == 0 {
@@ -116,7 +117,7 @@ impl Core {
 
         let page = self
             .db
-            .photos_page(offset, limit, kind, range)
+            .photos_page(offset, limit, kind, range, favorites)
             .map_err(CoreError::from)?;
         Ok(Some(page.into_iter().map(|p| self.photo_item(p)).collect()))
     }
@@ -139,9 +140,42 @@ impl Core {
             ratio: photo.ratio,
             no_thumb: photo.thumb_state == db::THUMB_NONE,
             kind: photo.kind,
+            favorite: photo.favorite,
         }
     }
 
+    /// Add or remove Proton's `Favorite` tag on a photo, then record the change
+    /// locally so the gallery reflects it without waiting for a timeline refresh.
+    ///
+    /// The SDK reports one outcome per photo rather than failing the call, so the
+    /// single outcome is unwrapped back into this request's result.
+    pub(crate) fn set_photo_favorite(&self, uid: &NodeUid, favorite: bool) -> CoreResult<()> {
+        let photos = self.photos();
+        let update = PhotoTagsUpdate {
+            node_uid: uid.clone(),
+            tags_to_add: if favorite {
+                vec![PhotoTag::Favorite]
+            } else {
+                Vec::new()
+            },
+            tags_to_remove: if favorite {
+                Vec::new()
+            } else {
+                vec![PhotoTag::Favorite]
+            },
+        };
+        let outcomes = self
+            .rt
+            .block_on(photos.update_photos(std::slice::from_ref(&update)))
+            .map_err(|e| CoreError::from_api(&e, "update photo tags"))?;
+        pdfs_core::batch::into_unit(outcomes)
+            .map_err(|e| CoreError::from_api(&e, "update photo tags"))?;
+        self.db
+            .photos_set_favorite(&uid.to_string(), favorite)
+            .map_err(CoreError::from)?;
+        Ok(())
+    }
+
     /// Thumbnails for `uids`, served from the cache, fetched from the server for
     /// whatever is missing, and — for the photos the server has no thumbnail for
     /// at all — *generated locally* from the full file (see
@@ -381,7 +415,7 @@ impl Core {
         // learned before (or classifies from nothing, i.e. a still photo), so a
         // partial resolve never blanks the timeline.
         let uids: Vec<NodeUid> = items.iter().map(|it| it.uid.clone()).collect();
-        let mut meta: HashMap<String, (Option<String>, Option<String>)> = HashMap::new();
+        let mut meta: HashMap<String, (Option<String>, Option<String>, bool)> = HashMap::new();
         for chunk in uids.chunks(TIMELINE_ENRICH_CHUNK) {
             match photos.enumerate_nodes(chunk).await {
                 Ok(nodes) => {
@@ -390,19 +424,35 @@ impl Core {
                             NodeKind::File { media_type, .. } => Some(media_type.clone()),
                             NodeKind::Folder => None,
                         };
-                        meta.insert(node.uid.to_string(), (Some(node.name), media_type));
+                        // The favourite state is a tag on the photo node, so it
+                        // rides along with the metadata resolve rather than
+                        // costing a listing of its own.
+                        let favorite = node
+                            .photo
+                            .as_ref()
+                            .is_some_and(|p| p.tags.contains(&PhotoTag::Favorite));
+                        meta.insert(
+                            node.uid.to_string(),
+                            (Some(node.name), media_type, favorite),
+                        );
                     }
                 }
                 Err(e) => warn!(error = %e, "resolving photo metadata for a timeline chunk failed"),
             }
         }
 
-        let rows: Vec<(String, i64, Option<String>, Option<String>)> = items
+        let rows: Vec<db::TimelineRow> = items
             .iter()
             .map(|it| {
                 let key = it.uid.to_string();
-                let (name, media_type) = meta.get(&key).cloned().unwrap_or((None, None));
-                (key, it.capture_time, name, media_type)
+                // An unresolved photo keeps whatever was learned before, for the
+                // favourite flag as much as for the name and media type.
+                match meta.get(&key).cloned() {
+                    Some((name, media_type, favorite)) => {
+                        (key, it.capture_time, name, media_type, Some(favorite))
+                    }
+                    None => (key, it.capture_time, None, None, None),
+                }
             })
             .collect();
         self.db.photos_replace(&rows).map_err(CoreError::from)?;

crates/pdfs-fuse/src/revisions.rs

@@ -0,0 +1,264 @@
+//! File version history: listing a file's revisions, restoring one, deleting
+//! one, and writing an old one out to a local path.
+//!
+//! Proton Drive keeps every revision a client ever committed, and until now the
+//! daemon only ever addressed the active one — a file that a sync pass or a
+//! `(sync-conflict)` fork overwrote could only be recovered from whatever the
+//! local `recovery/` directory happened to hold. This is the remote equivalent,
+//! and it is the reason a *restore* here is not a download-then-upload: the
+//! server re-points the file at an existing revision, so no content crosses the
+//! wire and nothing lands in the drain queue.
+//!
+//! Two things about the SDK surface leak into every method here:
+//!
+//! - **A restore is asynchronous.** The server answers 202 and swaps the active
+//!   revision in the background, so the daemon drops its cached content for the
+//!   file rather than trying to describe the new one. The next read repopulates
+//!   from whatever is live by then.
+//! - **The active revision cannot be deleted**, and the server — not this code —
+//!   is what enforces that. [`Core::delete_revision_for_uid`] refuses locally
+//!   too, so the user gets a sentence instead of an API code.
+
+use std::path::Path;
+
+use pdfs_core::control::{ActivityKind, RevisionInfo};
+use pdfs_core::{CoreError, CoreResult};
+use proton_drive_rs::Revision;
+use proton_drive_rs::proton_sdk::ids::NodeUid;
+use tracing::info;
+
+use super::Core;
+
+/// A [`Revision`] as the control protocol reports it.
+fn revision_info(revision: Revision) -> RevisionInfo {
+    RevisionInfo {
+        is_active: revision.is_active(),
+        id: revision.revision_id,
+        created: revision.creation_time,
+        size_on_storage: revision.size_on_storage,
+        claimed_size: revision.claimed_size,
+        claimed_modified: revision.claimed_modification_time,
+        // An empty signer means the node key signed it anonymously, which is not
+        // an address a front-end can show.
+        signed_by: revision.signature_email.filter(|email| !email.is_empty()),
+        has_thumbnails: revision.has_thumbnails,
+    }
+}
+
+impl Core {
+    /// The version history of the file at mountpoint-relative `rel`.
+    pub(crate) fn list_revisions(&self, rel: &Path) -> CoreResult<Vec<RevisionInfo>> {
+        let (_ino, uid) = self.resolve(rel)?;
+        self.list_revisions_for_uid(&uid)
+    }
+
+    /// [`Core::list_revisions`] for a node addressed through any daemon location.
+    pub(crate) fn list_revisions_by_uid(&self, uid: &str) -> CoreResult<Vec<RevisionInfo>> {
+        let uid = self.resolve_anywhere(uid)?;
+        self.list_revisions_for_uid(&uid)
+    }
+
+    fn list_revisions_for_uid(&self, uid: &NodeUid) -> CoreResult<Vec<RevisionInfo>> {
+        let revisions = self
+            .rt
+            .block_on(self.client.enumerate_revisions(uid))
+            .map_err(|e| CoreError::from_api(&e, "list revisions"))?;
+        Ok(revisions.into_iter().map(revision_info).collect())
+    }
+
+    /// Make `revision_id` the current content of the file at `rel`.
+    pub(crate) fn restore_revision(&self, rel: &Path, revision_id: &str) -> CoreResult<()> {
+        let (_ino, uid) = self.resolve(rel)?;
+        let name = rel
+            .file_name()
+            .map(|n| n.to_string_lossy().into_owned())
+            .unwrap_or_else(|| uid.to_string());
+        self.restore_revision_for_uid(&uid, revision_id, &name)?;
+        // The size and mtime a listing shows describe the revision that was
+        // active a moment ago.
+        self.invalidate_parent_listing(rel);
+        Ok(())
+    }
+
+    /// [`Core::restore_revision`] for a node addressed through any daemon
+    /// location.
+    pub(crate) fn restore_revision_by_uid(&self, uid: &str, revision_id: &str) -> CoreResult<()> {
+        let uid = self.resolve_anywhere(uid)?;
+        let name = uid.to_string();
+        self.restore_revision_for_uid(&uid, revision_id, &name)
+    }
+
+    fn restore_revision_for_uid(
+        &self,
+        uid: &NodeUid,
+        revision_id: &str,
+        name: &str,
+    ) -> CoreResult<()> {
+        // Restoring rewrites the file's content: the same authority a write
+        // needs, not the one a read needs.
+        self.require_uid_writable(uid)
+            .map_err(|error| self.errno_error(error, "restore revision"))?;
+        if let Err(e) = self
+            .rt
+            .block_on(self.client.restore_revision(uid, revision_id))
+        {
+            let error = CoreError::from_api(&e, "restore revision");
+            self.log_activity(ActivityKind::Restore, name, &error, false);
+            return Err(error);
+        }
+        // The active revision is changing behind our back — anything describing
+        // the old one (cached blocks, an open reader, the size in a listing) is
+        // stale from here on, and the server may not have applied the swap yet.
+        self.cache.evict(uid);
+        self.evict_reader(uid);
+        self.log_activity(
+            ActivityKind::Restore,
+            name,
+            format!("restored version {revision_id}"),
+            true,
+        );
+        info!(%uid, revision_id, name, "restored an earlier revision");
+        Ok(())
+    }
+
+    /// Permanently delete `revision_id` from the history of the file at `rel`.
+    pub(crate) fn delete_revision(&self, rel: &Path, revision_id: &str) -> CoreResult<()> {
+        let (_ino, uid) = self.resolve(rel)?;
+        let name = rel
+            .file_name()
+            .map(|n| n.to_string_lossy().into_owned())
+            .unwrap_or_else(|| uid.to_string());
+        self.delete_revision_for_uid(&uid, revision_id, &name)
+    }
+
+    /// [`Core::delete_revision`] for a node addressed through any daemon
+    /// location.
+    pub(crate) fn delete_revision_by_uid(&self, uid: &str, revision_id: &str) -> CoreResult<()> {
+        let uid = self.resolve_anywhere(uid)?;
+        let name = uid.to_string();
+        self.delete_revision_for_uid(&uid, revision_id, &name)
+    }
+
+    fn delete_revision_for_uid(
+        &self,
+        uid: &NodeUid,
+        revision_id: &str,
+        name: &str,
+    ) -> CoreResult<()> {
+        self.require_uid_writable(uid)
+            .map_err(|error| self.errno_error(error, "delete revision"))?;
+        // The server rejects deleting the active revision; saying so here is a
+        // sentence the user can act on instead of an API code, and it costs one
+        // listing the front-end already made to show the version list.
+        let revisions = self
+            .rt
+            .block_on(self.client.enumerate_revisions(uid))
+            .map_err(|e| CoreError::from_api(&e, "list revisions"))?;
+        let target = revisions
+            .iter()
+            .find(|r| r.revision_id == revision_id)
+            .ok_or_else(|| CoreError::not_found(format!("no version {revision_id} on {name}")))?;
+        if target.is_active() {
+            return Err(CoreError::invalid(format!(
+                "version {revision_id} is the current content of {name}; restore another version first"
+            )));
+        }
+        self.rt
+            .block_on(self.client.delete_revision(uid, revision_id))
+            .map_err(|e| CoreError::from_api(&e, "delete revision"))?;
+        self.log_activity(
+            ActivityKind::DeleteForever,
+            name,
+            format!("deleted version {revision_id}"),
+            true,
+        );
+        info!(%uid, revision_id, name, "deleted a revision");
+        Ok(())
+    }
+
+    /// Write `revision_id` of the file at `rel` to the absolute local path
+    /// `dest`, leaving the file itself untouched.
+    pub(crate) fn save_revision_as(
+        &self,
+        rel: &Path,
+        revision_id: &str,
+        dest: &Path,
+    ) -> CoreResult<u64> {
+        let (_ino, uid) = self.resolve(rel)?;
+        let name = rel
+            .file_name()
+            .map(|n| n.to_string_lossy().into_owned())
+            .unwrap_or_else(|| uid.to_string());
+        self.save_revision_for_uid(&uid, revision_id, &name, dest)
+    }
+
+    /// [`Core::save_revision_as`] for a node addressed through any daemon
+    /// location.
+    pub(crate) fn save_revision_as_by_uid(
+        &self,
+        uid: &str,
+        revision_id: &str,
+        dest: &Path,
+    ) -> CoreResult<u64> {
+        let uid = self.resolve_anywhere(uid)?;
+        let name = uid.to_string();
+        self.save_revision_for_uid(&uid, revision_id, &name, dest)
+    }
+
+    fn save_revision_for_uid(
+        &self,
+        uid: &NodeUid,
+        revision_id: &str,
+        name: &str,
+        dest: &Path,
+    ) -> CoreResult<u64> {
+        if !dest.is_absolute() {
+            return Err(CoreError::invalid(format!(
+                "destination must be an absolute path: {}",
+                dest.display()
+            )));
+        }
+        // Never silently replace something the user already has: an old version
+        // written over the wrong file is exactly the loss this feature exists to
+        // undo.
+        if dest.exists() {
+            return Err(CoreError::invalid(format!(
+                "{} already exists",
+                dest.display()
+            )));
+        }
+        let file = std::fs::File::create(dest)
+            .map_err(|e| CoreError::internal(format!("create {}: {e}", dest.display())))?;
+        let mut out = std::io::BufWriter::new(file);
+        let written = match self
+            .rt
+            .block_on(self.client.download_revision_to(uid, revision_id, &mut out))
+            .map_err(|e| CoreError::from_api(&e, "download revision"))
+            .and_then(|()| {
+                use std::io::Write as _;
+                out.flush()
+                    .map_err(|e| CoreError::internal(format!("write {}: {e}", dest.display())))?;
+                let written = std::fs::metadata(dest)
+                    .map(|m| m.len())
+                    .map_err(|e| CoreError::internal(format!("stat {}: {e}", dest.display())))?;
+                Ok(written)
+            }) {
+            Ok(written) => written,
+            Err(error) => {
+                // A half-written file looks like a successful export to every
+                // tool that opens it afterwards.
+                let _ = std::fs::remove_file(dest);
+                self.log_activity(ActivityKind::Download, name, &error, false);
+                return Err(error);
+            }
+        };
+        self.log_activity(
+            ActivityKind::Download,
+            name,
+            format!("saved version {revision_id} to {}", dest.display()),
+            true,
+        );
+        info!(%uid, revision_id, dest = %dest.display(), "wrote an earlier revision to disk");
+        Ok(written)
+    }
+}

crates/pdfs-fuse/src/sweep.rs

@@ -21,6 +21,7 @@
 use std::collections::HashMap;
 use std::time::Duration;
 
+use pdfs_core::batch;
 use pdfs_core::config::SweepMode;
 use pdfs_core::control::ActivityKind;
 use proton_drive_rs::proton_sdk::ids::NodeUid;
@@ -149,6 +150,7 @@ impl Core {
         match self
             .rt
             .block_on(self.client.trash_nodes(std::slice::from_ref(uid)))
+            .and_then(batch::into_unit)
         {
             Ok(()) => {}
             Err(e) if super::is_gone(&e) => {}

crates/pdfs-fuse/src/sync.rs

@@ -38,6 +38,7 @@ use tracing::{debug, info, warn};
 
 use crate::transfers::{CountingWriter, OwnedCountingReader};
 use crate::{Core, now_secs, parse_uid};
+use pdfs_core::batch;
 use pdfs_core::control::{ActivityKind, TransferDirection};
 use pdfs_core::db::{StoredSyncEntry, StoredSyncFolder};
 use pdfs_core::syncignore::IgnoreRules;
@@ -513,7 +514,11 @@ impl Core {
                 let _ = self.db.sync_entry_remove(folder_id, rel);
                 continue;
             };
-            if let Err(e) = self.rt.block_on(self.client.trash_nodes(&[uid])) {
+            if let Err(e) = self
+                .rt
+                .block_on(self.client.trash_nodes(&[uid]))
+                .and_then(batch::into_unit)
+            {
                 warn!(rel, error = %e, "sync: trash remote failed");
                 self.log_activity(ActivityKind::Trash, base_name(rel), e.to_string(), false);
                 outcome.errors += 1;
@@ -754,6 +759,7 @@ impl Core {
                     if let Err(e) = self
                         .rt
                         .block_on(self.client.trash_nodes(std::slice::from_ref(&remote.uid)))
+                        .and_then(batch::into_unit)
                     {
                         warn!(rel, error = %e, "sync: trash remote failed");
                         self.log_activity(
@@ -808,7 +814,11 @@ impl Core {
         }
         delete_remote_dirs.sort_by_key(|(p, _)| std::cmp::Reverse(p.matches('/').count()));
         for (rel, uid) in delete_remote_dirs {
-            if let Err(e) = self.rt.block_on(self.client.trash_nodes(&[uid])) {
+            if let Err(e) = self
+                .rt
+                .block_on(self.client.trash_nodes(&[uid]))
+                .and_then(batch::into_unit)
+            {
                 warn!(rel, error = %e, "sync: trash remote folder failed");
                 self.log_activity(ActivityKind::Trash, base_name(&rel), e.to_string(), false);
                 outcome.errors += 1;