feat(core,fuse,cli,gui): discard a queued change and undo it here

A change Drive will not take could be exported or retried, never let go
of. The Sync page's Discard button (with a confirmation) and
`pdfs sync discard <id>` drop it: a node made here goes, as if deleted
before it was sent; a node Drive has goes back to what Drive has, under
the same inode, with every change queued on it, while what is queued
below it stays. A change on the wire or an open file is refused.

crates/pdfs-cli/src/main.rs

@@ -686,6 +686,16 @@ enum SyncCmd {
         /// Where to save it: a new file, or an existing folder to save it in.
         dest: PathBuf,
     },
+    /// Drop a queued change (by id from `sync queue`) and undo it here: a new file
+    /// or folder goes, anything else goes back to what Proton Drive has. Its
+    /// content is lost unless saved with `sync export` first.
+    Discard {
+        /// Queue id.
+        id: i64,
+        /// Do not ask first.
+        #[arg(long)]
+        yes: bool,
+    },
     /// Retry a queued operation now (by id from `sync queue`), or every failed one.
     Retry {
         /// Queue id; omit to retry every failed operation.
@@ -1040,6 +1050,14 @@ fn cmd_sync(action: SyncCmd) -> Result<()> {
         SyncCmd::Issues => return cmd_sync_issues(),
         SyncCmd::Export { id, dest } => return cmd_sync_export(id, &dest),
         SyncCmd::Limit { up, down } => return cmd_sync_limit(up, down),
+        SyncCmd::Discard { id, yes } => {
+            if !yes {
+                confirm(&format!(
+                    "Discard queued change {id} and undo it here? Content not exported is lost."
+                ))?;
+            }
+            ok_or_bail(control_request(CtlRequest::DiscardPendingOp { id })?)?
+        }
         SyncCmd::Retry { id } => ok_or_bail(control_request(CtlRequest::RetryPendingOp { id })?)?,
         SyncCmd::Resume { folder: None } => {
             ok_or_bail(control_request(CtlRequest::SetSyncPaused {

crates/pdfs-core/src/control.rs

@@ -481,6 +481,11 @@ pub enum Request {
     /// [`Request::ListPendingOps`]) to `dest`, an absolute path that must not
     /// exist yet. The op stays queued. Replies with [`Response::Ok`].
     ExportPendingOp { id: i64, dest: String },
+    /// Drop a queued change (`id` from [`Request::ListPendingOps`]) and undo
+    /// it locally: a node made here goes, a node Drive has goes back to what
+    /// Drive has, along with every other change queued on it. Its content is
+    /// lost unless it was exported first. Replies with [`Response::Ok`].
+    DiscardPendingOp { id: i64 },
     /// List the `(sync-conflict …)` copies under My Files and in synced
     /// folders. Replies with
     /// [`Response::Conflicts`].
@@ -3354,6 +3359,10 @@ mod tests {
                 },
                 r#"{"ExportPendingOp":{"id":7,"dest":"/tmp/out.bin"}}"#,
             ),
+            (
+                Request::DiscardPendingOp { id: 7 },
+                r#"{"DiscardPendingOp":{"id":7}}"#,
+            ),
         ];
         for (request, wire) in cases {
             assert_eq!(serde_json::to_string(&request).unwrap(), wire);

crates/pdfs-core/src/db/ops.rs

@@ -908,6 +908,41 @@ impl Db {
         Ok(blobs)
     }
 
+    /// Drop the ops queued on the node `uid` names, and nothing else, returning
+    /// the staged blobs they held. `None`, dropping nothing, while one of them
+    /// is on the wire: it may land after the drop.
+    ///
+    /// For a change the user discards on a node Drive has. Unlike
+    /// [`Db::delete_ops_for_uid`], what is queued below the node stays: the
+    /// files made in a folder whose rename is discarded still go in it.
+    pub fn discard_ops_on_node(&self, uid: &str) -> Result<Option<Vec<String>>> {
+        let mut conn = self.conn.lock();
+        let tx = conn.transaction()?;
+        let lid = node_lid(&tx, uid)?;
+        let on = on_node(2);
+        let sending: bool = tx.query_row(
+            &format!("SELECT EXISTS (SELECT 1 FROM pending_op WHERE {on} AND claimed_at <> 0)"),
+            params![uid, lid],
+            |r| r.get(0),
+        )?;
+        if sending {
+            return Ok(None);
+        }
+        let blobs = {
+            let mut stmt = tx.prepare(&format!(
+                "SELECT blob_path FROM pending_op WHERE {on} AND blob_path IS NOT NULL"
+            ))?;
+            let rows = stmt.query_map(params![uid, lid], |r| r.get::<_, String>(0))?;
+            rows.collect::<std::result::Result<Vec<_>, _>>()?
+        };
+        tx.execute(
+            &format!("DELETE FROM pending_op WHERE {on}"),
+            params![uid, lid],
+        )?;
+        tx.commit()?;
+        Ok(Some(blobs))
+    }
+
     /// What [`Db::finish_create`] does to the node rows and pins when a create
     /// lands, without the op ([`adopt_placeholder_row_tx`]).
     #[cfg(test)]

crates/pdfs-core/src/db/tests.rs

@@ -2633,6 +2633,58 @@ fn failed_atomic_trash_insert_keeps_prior_revision_and_blob_ownership() {
     );
 }
 
+/// Discarding a change to a node Drive has drops that node's ops and nothing
+/// queued below it, and nothing at all while one of them is on the wire.
+#[test]
+fn discarding_a_nodes_change_keeps_what_is_queued_inside_it() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_nodes(&[
+        folder("root", None, "My Files"),
+        folder("dir", Some("root"), "dir"),
+        file("inner", "dir", "inner.bin", 1),
+    ])
+    .unwrap();
+    let op = |kind: &str, link: &str, blob: Option<&str>| PendingOp {
+        id: 0,
+        kind: kind.to_string(),
+        uid: uid(link).to_string(),
+        parent_uid: None,
+        name: Some("renamed".to_string()),
+        blob_path: blob.map(str::to_string),
+        meta_json: Some("{}".to_string()),
+        created_at: 1,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    };
+    db.enqueue_op(&op(OP_RENAME, "dir", None)).unwrap();
+    db.enqueue_op(&op(OP_REVISION, "inner", Some("/staging/inner")))
+        .unwrap();
+
+    let dir = uid("dir").to_string();
+    assert_eq!(db.discard_ops_on_node(&dir).unwrap(), Some(vec![]));
+    let left: Vec<String> = db
+        .pending_ops()
+        .unwrap()
+        .into_iter()
+        .map(|op| op.uid)
+        .collect();
+    assert_eq!(left, vec![uid("inner").to_string()]);
+
+    let inner = uid("inner").to_string();
+    let claimed = db.claim_next_due_op(2).unwrap().unwrap();
+    assert_eq!(claimed.uid, inner);
+    assert_eq!(db.discard_ops_on_node(&inner).unwrap(), None);
+    assert_eq!(db.pending_ops().unwrap().len(), 1);
+    db.record_op_failure(claimed.id, "no answer", 0).unwrap();
+    db.release_op_claim(claimed.id).unwrap();
+    assert_eq!(
+        db.discard_ops_on_node(&inner).unwrap(),
+        Some(vec!["/staging/inner".to_string()])
+    );
+    assert!(db.pending_ops().unwrap().is_empty());
+}
+
 /// A revision op names no parent, so the queue alone cannot see that its file
 /// sits inside a folder being removed. The node tree can: trashing the folder
 /// has to take the uploads of the files below it along, or they fail forever

crates/pdfs-fuse/src/control.rs

@@ -377,6 +377,7 @@ fn touched_topics(request: &CtlRequest) -> &'static [Topic] {
         | R::RestoreSyncFolders { .. }
         | R::RestoreDeviceFolders { .. } => &[Topic::Locations, Topic::Devices],
         R::RetryPendingOp { .. } => &[Topic::Queue],
+        R::DiscardPendingOp { .. } => &[Topic::Queue, Topic::Files],
         R::ResolveConflict { .. } => &[Topic::Files, Topic::Trash, Topic::Conflicts],
         R::ShareNode { .. }
         | R::ShareNodeByUid { .. }
@@ -1412,6 +1413,12 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
                 Err(e) => CtlResponse::error(e),
             }
         }
+        Ok(CtlRequest::DiscardPendingOp { id }) => match core.discard_pending_op(id) {
+            Ok(path) => CtlResponse::Ok {
+                message: format!("discarded the queued change to {path}"),
+            },
+            Err(e) => CtlResponse::error(e),
+        },
         Ok(CtlRequest::RetryPendingOp { id }) => match core.retry_pending_ops(id) {
             Ok(0) if id.is_some() => CtlResponse::Ok {
                 message: "nothing to retry: the operation is already due or waiting for a rename"

crates/pdfs-fuse/src/queue.rs

@@ -2,8 +2,10 @@
 //! yet, why, and a way to stop waiting out a backoff.
 //!
 //! Read-only apart from "retry now", which only moves a row's next-attempt time,
-//! and "export", which copies a staged blob out. Nothing here drops a queued op:
-//! its staged blob is the only copy of the user's write (see `docs/RECOVERY.md`).
+//! "export", which copies a staged blob out, and "discard". Only a discard the
+//! user asks for drops a queued op: its staged blob is the only copy of the
+//! user's write (see `docs/RECOVERY.md`), so it is the user's call, made after
+//! an export or for a change with no content.
 
 use super::*;
 use std::io::Write;
@@ -91,6 +93,134 @@ impl Core {
         info!(id, dest = %dest.display(), written, "exported a queued change's content");
         Ok(written)
     }
+
+    /// Drop queued op `id` and undo the local change it stood for. Returns the
+    /// node's path.
+    ///
+    /// A node only made here goes, as if it had been deleted before it was
+    /// sent. A node Drive has goes back to what Drive has: every change queued
+    /// on it goes, since each was made on top of the one before. What is queued
+    /// below it stays.
+    pub(crate) fn discard_pending_op(&self, id: i64) -> CoreResult<String> {
+        let ops = self
+            .db
+            .pending_ops()
+            .map_err(|e| CoreError::internal(format!("listing the queue: {e}")))?;
+        let op = ops
+            .into_iter()
+            .find(|op| op.id == id)
+            .ok_or_else(|| CoreError::not_found(format!("no queued change {id}")))?;
+        let uid = parse_node_uid(&op.uid)
+            .ok_or_else(|| CoreError::internal(format!("queued change {id} names {}", op.uid)))?;
+        let path = self
+            .db
+            .node_path(&op.uid)
+            .ok()
+            .flatten()
+            .filter(|path| !path.is_empty())
+            .or(op.name)
+            .unwrap_or_else(|| op.uid.clone());
+        if self.is_open_anywhere(&uid) {
+            return Err(CoreError::conflict(format!(
+                "{path} is open; close it first"
+            )));
+        }
+        if is_local_uid(&uid) {
+            self.discard_local_node(&uid, &path)?;
+        } else {
+            self.discard_remote_change(&uid, &path)?;
+        }
+        info!(id, %uid, path, "discarded a queued change");
+        Ok(path)
+    }
+
+    /// Forget a node made here, and the ops that would have sent it.
+    fn discard_local_node(&self, uid: &NodeUid, path: &str) -> CoreResult<()> {
+        // Its children's ops go with its own, and their content with them.
+        if self
+            .db
+            .has_children(uid)
+            .map_err(|e| CoreError::internal(format!("listing {path}: {e}")))?
+        {
+            return Err(CoreError::conflict(format!(
+                "{path} is not empty; discard what is in it first"
+            )));
+        }
+        self.discard_queued_ops(uid)
+            .map_err(|_| CoreError::internal(format!("dropping the changes to {path}")))?;
+        self.for_each_mount(|st, notify| {
+            background::forget_and_notify(st, notify, uid);
+        });
+        self.cache.evict(uid);
+        self.evict_reader(uid);
+        Ok(())
+    }
+
+    /// Put a node Drive has back as Drive has it.
+    fn discard_remote_change(&self, uid: &NodeUid, path: &str) -> CoreResult<()> {
+        // Asked first: the change is only dropped once what replaces it is known.
+        let node = self
+            .fetch_node_remote(uid)
+            .map_err(|e| CoreError::from_api(&e, "reading the node from Drive"))?
+            .filter(|node| !node.trashed);
+        self.cancel_upload(uid);
+        let blobs = self
+            .db
+            .discard_ops_on_node(&uid.to_string())
+            .map_err(|e| CoreError::internal(format!("dropping the changes to {path}: {e}")))?
+            .ok_or_else(|| {
+                CoreError::conflict(format!("{path} is being sent right now; try again"))
+            })?;
+        self.release_dropped_ops(uid, blobs);
+        self.hidden.lock().remove(uid);
+        match node {
+            Some(node) => self.for_each_mount(|st, notify| {
+                put_back(st, notify, node);
+            }),
+            None => self.for_each_mount(|st, notify| {
+                background::forget_and_notify(st, notify, uid);
+            }),
+        }
+        self.cache.evict(uid);
+        self.evict_reader(uid);
+        Ok(())
+    }
+}
+
+/// Show `node` as Drive has it, where Drive has it, under the inode it has.
+fn put_back(st: &mut State, notify: &mut NotifyBatch, node: Node) {
+    let parent = node.parent_uid.as_ref().and_then(|parent| {
+        st.by_uid
+            .get(parent)
+            .copied()
+            .map(|ino| (ino, parent.clone()))
+    });
+    let here = st.by_uid.get(&node.uid).and_then(|ino| {
+        st.entries
+            .get(ino)
+            .map(|entry| (*ino, entry.parent, entry.node.name.clone()))
+    });
+    match (here, parent) {
+        (Some((ino, from, name)), Some((to, to_uid))) => {
+            st.relocate(ino, from, to, &to_uid, &node.name);
+            st.write_through(node.clone());
+            st.intern_mem(to, node);
+            notify.inval_entry(from, name);
+            notify.inval_inode(ino);
+        }
+        (here, parent) => {
+            if let Some((_, from, name)) = here {
+                st.forget_mem(&node.uid);
+                st.invalidate_listing(from);
+                notify.inval_entry(from, name);
+            }
+            if let Some((to, _)) = parent {
+                st.invalidate_listing(to);
+                notify.inval_entry(to, node.name.clone());
+            }
+            st.write_through(node);
+        }
+    }
 }
 
 /// Copy `blob` to `dest`, which must not exist: never replace something the

crates/pdfs-fuse/src/sim/daemon.rs

@@ -645,6 +645,59 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    #[test]
+    #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
+    fn a_discarded_change_is_undone_here_and_never_sent() {
+        let drive = FakeDrive::new();
+        for name in ["keep.txt", "edit.txt", "gone.txt"] {
+            drive.device().write(name, b"remote");
+        }
+        let before = drive.tree();
+        let dir = scratch("discard");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let mnt = daemon.mountpoint.clone();
+        let paused = |paused| Request::SetSyncPaused {
+            paused,
+            until: None,
+        };
+        daemon.request(&paused(true)).unwrap();
+        std::fs::rename(mnt.join("keep.txt"), mnt.join("moved.txt")).unwrap();
+        std::fs::write(mnt.join("edit.txt"), b"local").unwrap();
+        std::fs::remove_file(mnt.join("gone.txt")).unwrap();
+        std::fs::write(mnt.join("new.txt"), b"new").unwrap();
+        std::fs::create_dir(mnt.join("dir")).unwrap();
+
+        // Discarding one change drops the others queued on its node.
+        for _ in 0..10 {
+            let Some(op) = daemon.pending().unwrap().into_iter().next() else {
+                break;
+            };
+            let reply = daemon.request(&Request::DiscardPendingOp { id: op.id });
+            assert!(
+                matches!(reply, Ok(Response::Ok { .. })),
+                "{op:?}: {reply:?}"
+            );
+        }
+        assert!(daemon.pending().unwrap().is_empty());
+        let mut names: Vec<_> = std::fs::read_dir(&mnt)
+            .unwrap()
+            .map(|entry| entry.unwrap().file_name().into_string().unwrap())
+            .filter(|name| name.ends_with(".txt") || name == "dir")
+            .collect();
+        names.sort();
+        assert_eq!(names, ["edit.txt", "gone.txt", "keep.txt"]);
+        for name in &names {
+            assert_eq!(std::fs::read(mnt.join(name)).unwrap(), b"remote", "{name}");
+        }
+
+        daemon.request(&paused(false)).unwrap();
+        std::thread::sleep(Duration::from_secs(2));
+        assert_eq!(drive.tree(), before);
+
+        assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown)));
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     #[test]
     #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
     fn a_folder_whose_mkdir_is_queued_lists_after_a_refresh() {

crates/pdfs-gui/src/app/pages/locations.rs

@@ -708,6 +708,20 @@ fn queue_row(ui: &Rc<Ui>, op: &PendingOpInfo, now: i64) -> adw::ActionRow {
         export.connect_clicked(move |_| prompt_export_queued(&ui, id, &name));
         row.add_suffix(&export);
     }
+    if op.failing {
+        let discard = gtk4::Button::builder()
+            .label(gettext("Discard"))
+            .tooltip_text(gettext("Undo this change here instead of sending it"))
+            .valign(gtk4::Align::Center)
+            .build();
+        discard.add_css_class("flat");
+        let ui = ui.clone();
+        let id = op.id;
+        let path = op.path.clone();
+        let exportable = op.exportable;
+        discard.connect_clicked(move |_| prompt_discard_queued(&ui, id, &path, exportable));
+        row.add_suffix(&discard);
+    }
     row
 }
 
@@ -773,6 +787,62 @@ fn prompt_export_queued(ui: &Rc<Ui>, id: i64, name: &str) {
     });
 }
 
+/// Confirm, then have the daemon drop a queued change and undo it here.
+fn prompt_discard_queued(ui: &Rc<Ui>, id: i64, path: &str, exportable: bool) {
+    let body = if exportable {
+        // Translators: {path} is the file whose queued change would be dropped.
+        gettext_f(
+            "Your changes to {path} are undone here and never reach Proton Drive. Export the file first to keep your version.",
+            &[("path", path)],
+        )
+    } else {
+        // Translators: {path} is the file or folder whose queued change would be dropped.
+        gettext_f(
+            "The change to {path} is undone here and never reaches Proton Drive.",
+            &[("path", path)],
+        )
+    };
+    let dialog = adw::AlertDialog::builder()
+        .heading(gettext("Discard Change?"))
+        .body(body)
+        .build();
+    dialog.add_response("cancel", &gettext("Cancel"));
+    dialog.add_response("discard", &gettext("Discard"));
+    dialog.set_response_appearance("discard", adw::ResponseAppearance::Destructive);
+    dialog.set_default_response(Some("cancel"));
+    dialog.set_close_response("cancel");
+    let window = ui_window(ui);
+    let ui = ui.clone();
+    dialog.connect_response(None, move |_, resp| {
+        if resp == "discard" {
+            discard_queued(&ui, id);
+        }
+    });
+    dialog.present(window.as_ref());
+}
+
+fn discard_queued(ui: &Rc<Ui>, id: i64) {
+    let rx = spawn_request(ui.dirs.control_socket(), Request::DiscardPendingOp { id });
+    let ui = ui.clone();
+    glib::spawn_future_local(async move {
+        match rx.recv().await {
+            Ok(Ok(Response::Ok { .. })) => {
+                toast(&ui, &gettext("Change discarded"));
+                ui.locations.queue.painted.borrow_mut().clear();
+                refresh_queue(&ui);
+            }
+            Ok(Ok(Response::Error { message, kind })) => {
+                toast_failure(&ui, &gettext("Couldn't discard the change"), &message, kind)
+            }
+            _ => toast_error(
+                &ui,
+                &gettext("Couldn't discard the change"),
+                &gettext("The Proton Drive service didn't respond."),
+            ),
+        }
+    });
+}
+
 /// Ask the daemon to stop waiting out one op's backoff, or every failed op's.
 fn retry_queued(ui: &Rc<Ui>, id: Option<i64>) {
     let rx = spawn_request(ui.dirs.control_socket(), Request::RetryPendingOp { id });