feat(core,fuse,cli,gui): report changes Drive refuses as sync issues

A change Drive refuses because the storage is full, access was taken away,
its folder is gone, a folder limit is reached or the change was rejected
used to look like any other failure until six attempts had passed. The
drain now sorts the refusal into an issue and stores it on the queued op
(schema 38), so it needs attention at once in the app, tray and CLI. The
change stays queued and is retried; nothing is deleted.

The queue listing carries the issue and whether the op has file content.
ExportPendingOp copies that content to a new file: the Sync page offers it
as an Export button next to each issue, and the CLI as
`pdfs sync export <id> <dest>`. `pdfs sync issues` lists what needs
attention, and `pdfs sync queue` shows the issue per change.

docs/MILESTONE-3.0.0-STATUS.md tracks what is left for 3.0.0.

crates/pdfs-fuse/src/queue.rs

@@ -1,12 +1,14 @@
 //! The user's view of the pending-op queue: what has not reached the remote
 //! 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.
-//! Nothing here drops a queued op: its staged blob is the only copy of the
-//! user's write (see `docs/RECOVERY.md`).
+//! 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`).
 
 use super::*;
-use pdfs_core::control::PendingOpInfo;
+use std::io::Write;
+
+use pdfs_core::control::{PendingOpInfo, SyncIssue};
 use pdfs_core::db::{FAILING_ATTEMPTS, PARK_UNTIL};
 
 impl Core {
@@ -16,10 +18,15 @@ impl Core {
             .db
             .pending_ops()
             .map_err(|e| CoreError::internal(format!("listing the queue: {e}")))?;
+        let issues = self
+            .db
+            .op_issues()
+            .map_err(|e| CoreError::internal(format!("listing sync issues: {e}")))?;
         Ok(ops
             .into_iter()
             .map(|op| {
                 let parked = op.next_attempt_at >= PARK_UNTIL;
+                let exportable = exportable(&op).is_some();
                 let path = self
                     .db
                     .node_path(&op.uid)
@@ -37,7 +44,9 @@ impl Core {
                     queued_at: op.created_at / 1000,
                     next_attempt_at: (!parked).then_some(op.next_attempt_at / 1000),
                     parked,
-                    failing: op.attempts >= FAILING_ATTEMPTS,
+                    failing: op.attempts >= FAILING_ATTEMPTS || issues.contains_key(&op.id),
+                    issue: issues.get(&op.id).map(|issue| SyncIssue::parse(issue)),
+                    exportable,
                 }
             })
             .collect())
@@ -57,4 +66,107 @@ impl Core {
         }
         Ok(moved)
     }
+
+    /// Copy the content of queued op `id` to `dest`, which must not exist yet.
+    /// The op stays queued. Returns the bytes written.
+    pub(crate) fn export_pending_op(&self, id: i64, dest: &Path) -> CoreResult<u64> {
+        if !dest.is_absolute() {
+            return Err(CoreError::invalid(format!(
+                "destination must be an absolute path: {}",
+                dest.display()
+            )));
+        }
+        let ops = self
+            .db
+            .pending_ops()
+            .map_err(|e| CoreError::internal(format!("listing the queue: {e}")))?;
+        let op = ops
+            .iter()
+            .find(|op| op.id == id)
+            .ok_or_else(|| CoreError::not_found(format!("no queued change {id}")))?;
+        let blob = exportable(op).ok_or_else(|| {
+            CoreError::invalid(format!("queued change {id} carries no file content"))
+        })?;
+        let written = copy_to_new_file(blob, dest)?;
+        info!(id, dest = %dest.display(), written, "exported a queued change's content");
+        Ok(written)
+    }
+}
+
+/// Copy `blob` to `dest`, which must not exist: never replace something the
+/// user already has.
+fn copy_to_new_file(blob: &Path, dest: &Path) -> CoreResult<u64> {
+    let mut source = File::open(blob)
+        .map_err(|e| CoreError::internal(format!("open {}: {e}", blob.display())))?;
+    let mut out = std::fs::OpenOptions::new()
+        .write(true)
+        .create_new(true)
+        .open(dest)
+        .map_err(|e| match e.kind() {
+            std::io::ErrorKind::AlreadyExists => {
+                CoreError::invalid(format!("{} already exists", dest.display()))
+            }
+            _ => CoreError::internal(format!("create {}: {e}", dest.display())),
+        })?;
+    std::io::copy(&mut source, &mut out)
+        .and_then(|written| out.flush().and(out.sync_all()).map(|()| written))
+        .map_err(|e| CoreError::internal(format!("write {}: {e}", dest.display())))
+}
+
+/// The staged blob of an op that carries file content.
+fn exportable(op: &PendingOp) -> Option<&Path> {
+    matches!(op.kind.as_str(), OP_CREATE | OP_REVISION)
+        .then_some(op.blob_path.as_deref())
+        .flatten()
+        .map(Path::new)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn op(kind: &str, blob: Option<&str>) -> PendingOp {
+        PendingOp {
+            id: 1,
+            kind: kind.to_string(),
+            uid: "vol~a".to_string(),
+            parent_uid: None,
+            name: None,
+            blob_path: blob.map(str::to_string),
+            meta_json: None,
+            created_at: 0,
+            attempts: 0,
+            last_error: None,
+            next_attempt_at: 0,
+        }
+    }
+
+    #[test]
+    fn only_a_change_that_carries_file_content_can_be_exported() {
+        assert!(exportable(&op(OP_CREATE, Some("/staging/a"))).is_some());
+        assert!(exportable(&op(OP_REVISION, Some("/staging/a"))).is_some());
+        assert!(exportable(&op(OP_CREATE, None)).is_none());
+        assert!(exportable(&op(OP_RENAME, None)).is_none());
+        assert!(exportable(&op(OP_TRASH, None)).is_none());
+    }
+
+    #[test]
+    fn an_export_copies_the_bytes_and_never_replaces_a_file() {
+        let dir = std::env::temp_dir().join(format!("pdfs-export-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let blob = dir.join("blob");
+        std::fs::write(&blob, b"queued bytes").unwrap();
+
+        let dest = dir.join("copy.txt");
+        assert_eq!(copy_to_new_file(&blob, &dest).unwrap(), 12);
+        assert_eq!(std::fs::read(&dest).unwrap(), b"queued bytes");
+
+        std::fs::write(&dest, b"the user's own").unwrap();
+        assert!(copy_to_new_file(&blob, &dest).is_err());
+        assert_eq!(std::fs::read(&dest).unwrap(), b"the user's own");
+        // The staged blob is the only copy of the change: an export leaves it.
+        assert_eq!(std::fs::read(&blob).unwrap(), b"queued bytes");
+
+        let _ = std::fs::remove_dir_all(&dir);
+    }
 }