fix(core,fuse,cli): harden writes, trash and cleanup found by account acceptance runs

The FUSE acceptance suite now runs on the signed-in account by default and
cleans up after itself; its first live runs found three daemon bugs.

- B101: a pwrite past EOF now authors the hole it leaves, so truncated bytes
  no longer come back from the base. A write attached to a create while that
  create uploads becomes a revision of the real uid instead of being dropped.
- B102: removing a folder drops the queued revisions of files below it, found
  through the node tree, with their blobs and in-flight uploads. `pdfs rm`
  now drops queued ops too.
- B103: each server call of a trash refresh times out after 120 s, so one hung
  call can no longer freeze the trash listing until restart.

Acceptance suite: account mode with a manifest, lock and reaper; cleanup by
uid; queued ops naming a test root count as leftovers; new edge-case,
throughput and B100/B101 cases; the wide-directory case is split from the
deep tree and gets a doubled budget. Adds `pdfs trash --json`.

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

@@ -66,6 +66,52 @@ pub fn op_supersedes(kind: &str) -> bool {
     matches!(kind, OP_REVISION | OP_RENAME | OP_TRASH)
 }
 
+/// Delete the ops that removing `uid` takes with it, returning the staged blobs
+/// they held.
+///
+/// That is every op for `uid` or for anything queued beneath it, plus the
+/// revisions of files the node tree places beneath it. A revision op carries
+/// no `parent_uid`, so the queue alone cannot tell that its file sits inside
+/// a removed folder: the upload outlived the folder, failed forever against a
+/// trashed node, and kept its blob on disk (`docs/BUGS.md` B102).
+///
+/// Only revisions are taken from the node tree. A queued rename that moved a
+/// node *into* the folder has to land, or the node would stay where it was
+/// on the server instead of going to the trash with the folder.
+fn drop_doomed_ops(tx: &rusqlite::Transaction<'_>, uid: &str) -> Result<Vec<String>> {
+    const DOOMED: &str = "
+        WITH RECURSIVE
+          queued(uid) AS (
+            SELECT ?1
+            UNION
+            SELECT p.uid FROM pending_op p JOIN queued q ON p.parent_uid = q.uid
+          ),
+          below(uid) AS (
+            SELECT ?1
+            UNION
+            SELECT n.uid FROM nodes n JOIN below b ON n.parent_uid = b.uid
+          ),
+          doomed(id) AS (
+            SELECT id FROM pending_op
+            WHERE uid IN (SELECT uid FROM queued)
+               OR (kind = 'revision' AND uid IN (SELECT uid FROM below))
+          )";
+    let blobs: Vec<String> = {
+        let mut stmt = tx.prepare(&format!(
+            "{DOOMED}
+             SELECT blob_path FROM pending_op
+             WHERE id IN (SELECT id FROM doomed) AND blob_path IS NOT NULL"
+        ))?;
+        let rows = stmt.query_map(params![uid], |r| r.get::<_, String>(0))?;
+        rows.collect::<std::result::Result<Vec<_>, _>>()?
+    };
+    tx.execute(
+        &format!("{DOOMED} DELETE FROM pending_op WHERE id IN (SELECT id FROM doomed)"),
+        params![uid],
+    )?;
+    Ok(blobs)
+}
+
 /// The volume id given to a node that exists only on this machine, so far. A
 /// real [`NodeUid`] is `{volume}~{link}`, so a placeholder is `local~<uuid>` and
 /// round-trips through the same `Display`/parse path as any other uid.
@@ -252,27 +298,9 @@ impl Db {
         name: &str,
         created_at: i64,
     ) -> Result<(i64, Vec<String>)> {
-        const SUBTREE: &str = "
-            WITH RECURSIVE doomed(uid) AS (
-              SELECT ?1
-              UNION
-              SELECT p.uid FROM pending_op p JOIN doomed d ON p.parent_uid = d.uid
-            )";
         let mut conn = self.conn.lock();
         let tx = conn.transaction()?;
-        let blobs: Vec<String> = {
-            let mut stmt = tx.prepare(&format!(
-                "{SUBTREE}
-                 SELECT blob_path FROM pending_op
-                 WHERE uid IN (SELECT uid FROM doomed) AND blob_path IS NOT NULL"
-            ))?;
-            let rows = stmt.query_map(params![uid], |row| row.get::<_, String>(0))?;
-            rows.collect::<std::result::Result<Vec<_>, _>>()?
-        };
-        tx.execute(
-            &format!("{SUBTREE} DELETE FROM pending_op WHERE uid IN (SELECT uid FROM doomed)"),
-            params![uid],
-        )?;
+        let blobs = drop_doomed_ops(&tx, uid)?;
         tx.execute(
             "INSERT INTO pending_op
                (kind, uid, parent_uid, name, blob_path, meta_json, created_at, next_attempt_at)
@@ -326,6 +354,67 @@ impl Db {
         Ok(Some(AttachedBlob { id, superseded }))
     }
 
+    /// Retire a queued create that has just landed as `real`, and repoint
+    /// everything that named its placeholder, in one transaction.
+    ///
+    /// The drain uploads the blob the create held when it was picked up. A
+    /// write that closes while that upload is on the wire attaches a newer blob
+    /// to the *same row* (`attach_blob_to_create`), so deleting the row by id
+    /// threw that write away: the server kept the older bytes and nothing was
+    /// left queued to replace them. When the row no longer holds `uploaded`, it
+    /// becomes a revision of `real` instead, with its sidecar passed through
+    /// `rewrite` so it names the real node. Returns that blob and sidecar, or
+    /// `None` when the create is simply done.
+    pub fn finish_create(
+        &self,
+        id: i64,
+        uploaded: Option<&str>,
+        local: &str,
+        real: &str,
+        rewrite: impl FnOnce(&str) -> Option<String>,
+    ) -> Result<Option<(String, String)>> {
+        let mut conn = self.conn.lock();
+        let tx = conn.transaction()?;
+        let row: Option<(Option<String>, Option<String>)> = tx
+            .query_row(
+                "SELECT blob_path, meta_json FROM pending_op WHERE id = ?1",
+                params![id],
+                |r| Ok((r.get(0)?, r.get(1)?)),
+            )
+            .optional()?;
+        let newer = match row {
+            Some((Some(blob), Some(meta))) if uploaded != Some(blob.as_str()) => {
+                rewrite(&meta).map(|meta| (blob, meta))
+            }
+            _ => None,
+        };
+        match &newer {
+            Some((_, meta)) => {
+                tx.execute(
+                    "UPDATE pending_op
+                     SET kind = ?2, uid = ?3, parent_uid = NULL, name = NULL, meta_json = ?4,
+                         attempts = 0, last_error = NULL, next_attempt_at = 0
+                     WHERE id = ?1",
+                    params![id, OP_REVISION, real, meta],
+                )?;
+            }
+            None => {
+                tx.execute("DELETE FROM pending_op WHERE id = ?1", params![id])?;
+            }
+        }
+        tx.execute(
+            "UPDATE pending_op SET parent_uid = ?2 WHERE parent_uid = ?1",
+            params![local, real],
+        )?;
+        tx.execute(
+            "UPDATE nodes SET parent_uid = ?2 WHERE parent_uid = ?1",
+            params![local, real],
+        )?;
+        tx.execute("DELETE FROM nodes WHERE uid = ?1", params![local])?;
+        tx.commit()?;
+        Ok(newer)
+    }
+
     /// Replace the sidecar of a queued op, for a baseline that has moved under
     /// it.
     ///
@@ -428,29 +517,12 @@ impl Db {
     /// and in the user's pending count, for the life of the database.
     ///
     /// Only `create`/`mkdir` ops carry a `parent_uid`, so for a file this
-    /// recursion finds nothing and costs one query.
+    /// recursion finds nothing and costs one query. Revisions of files the
+    /// node tree places beneath `uid` go too; see [`drop_doomed_ops`].
     pub fn delete_ops_for_uid(&self, uid: &str) -> Result<Vec<String>> {
-        const SUBTREE: &str = "
-            WITH RECURSIVE doomed(uid) AS (
-              SELECT ?1
-              UNION
-              SELECT p.uid FROM pending_op p JOIN doomed d ON p.parent_uid = d.uid
-            )";
         let mut conn = self.conn.lock();
         let tx = conn.transaction()?;
-        let blobs: Vec<String> = {
-            let mut stmt = tx.prepare(&format!(
-                "{SUBTREE}
-                 SELECT blob_path FROM pending_op
-                 WHERE uid IN (SELECT uid FROM doomed) AND blob_path IS NOT NULL"
-            ))?;
-            let rows = stmt.query_map(params![uid], |r| r.get::<_, String>(0))?;
-            rows.collect::<std::result::Result<Vec<_>, _>>()?
-        };
-        tx.execute(
-            &format!("{SUBTREE} DELETE FROM pending_op WHERE uid IN (SELECT uid FROM doomed)"),
-            params![uid],
-        )?;
+        let blobs = drop_doomed_ops(&tx, uid)?;
         tx.commit()?;
         Ok(blobs)
     }

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

@@ -2498,6 +2498,71 @@ fn failed_atomic_trash_insert_keeps_prior_revision_and_blob_ownership() {
     );
 }
 
+/// 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
+/// against a trashed node and keep their bytes on disk (B102). A rename that
+/// moved a node into the folder still has to land.
+#[test]
+fn trashing_a_folder_drops_the_queued_revisions_of_files_below_it() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_nodes(&[
+        folder("root", None, "My Files"),
+        folder("doomed", Some("root"), "doomed"),
+        folder("sub", Some("doomed"), "sub"),
+        file("deep", "sub", "deep.bin", 1),
+        file("moved", "doomed", "moved.txt", 1),
+        file("kept", "root", "kept.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: None,
+        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_REVISION, "deep", Some("/staging/deep")))
+        .unwrap();
+    db.enqueue_op(&op(OP_RENAME, "moved", None)).unwrap();
+    db.enqueue_op(&op(OP_REVISION, "kept", Some("/staging/kept")))
+        .unwrap();
+
+    let doomed = uid("doomed").to_string();
+    let (_, blobs) = db.replace_ops_with_trash(&doomed, "doomed", 2).unwrap();
+    assert_eq!(blobs, vec!["/staging/deep"]);
+
+    let mut left: Vec<(String, String)> = db
+        .pending_ops()
+        .unwrap()
+        .into_iter()
+        .map(|op| (op.kind, op.uid))
+        .collect();
+    left.sort();
+    assert_eq!(
+        left,
+        vec![
+            (OP_RENAME.to_string(), uid("moved").to_string()),
+            (OP_REVISION.to_string(), uid("kept").to_string()),
+            (OP_TRASH.to_string(), doomed.clone()),
+        ]
+    );
+
+    // Deleting outright follows the same tree.
+    db.enqueue_op(&op(OP_REVISION, "deep", Some("/staging/deep2")))
+        .unwrap();
+    assert_eq!(
+        db.delete_ops_for_uid(&uid("sub").to_string()).unwrap(),
+        vec!["/staging/deep2"]
+    );
+}
+
 /// Deleting a folder that was created offline must take the ops queued
 /// underneath it with it: they name a placeholder parent that will now never
 /// become real, so nothing could ever drain them and nothing is left to
@@ -2699,6 +2764,99 @@ fn attaching_to_an_already_drained_create_reports_it_is_gone() {
     assert!(out.is_none());
 }
 
+#[test]
+fn a_write_attached_while_its_create_uploads_becomes_a_revision() {
+    let db = Db::open_in_memory().unwrap();
+    let local = "local~abc";
+    let real = uid("real").to_string();
+    let create = |blob: Option<&str>| PendingOp {
+        id: 0,
+        kind: OP_CREATE.to_string(),
+        uid: local.to_string(),
+        parent_uid: Some(uid("parent").to_string()),
+        name: Some("new.txt".to_string()),
+        blob_path: blob.map(str::to_string),
+        meta_json: blob.map(|_| "{\"uid\":\"local~abc\"}".to_string()),
+        created_at: 1,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    };
+    let (id, _) = db.enqueue_op(&create(None)).unwrap();
+    // The drain picked the create up with no blob; the file's first write
+    // closes while the empty upload is on the wire.
+    db.attach_blob_to_create(local, "/staging/written", "{\"uid\":\"local~abc\"}")
+        .unwrap()
+        .expect("create is still queued");
+
+    let newer = db
+        .finish_create(id, None, local, &real, |json| {
+            Some(json.replace("local~abc", "vol~real"))
+        })
+        .unwrap();
+
+    assert_eq!(
+        newer,
+        Some((
+            "/staging/written".to_string(),
+            "{\"uid\":\"vol~real\"}".to_string()
+        ))
+    );
+    let ops = db.pending_ops().unwrap();
+    assert_eq!(ops.len(), 1, "the write is still queued");
+    assert_eq!(ops[0].kind, OP_REVISION);
+    assert_eq!(ops[0].uid, real);
+    assert_eq!(ops[0].blob_path.as_deref(), Some("/staging/written"));
+    assert_eq!(ops[0].parent_uid, None);
+}
+
+#[test]
+fn a_create_whose_blob_landed_is_retired() {
+    let db = Db::open_in_memory().unwrap();
+    let local = "local~abc";
+    let real = uid("real").to_string();
+    let (id, _) = db
+        .enqueue_op(&PendingOp {
+            id: 0,
+            kind: OP_CREATE.to_string(),
+            uid: local.to_string(),
+            parent_uid: Some(uid("parent").to_string()),
+            name: Some("new.txt".to_string()),
+            blob_path: Some("/staging/sent".to_string()),
+            meta_json: Some("{}".to_string()),
+            created_at: 1,
+            attempts: 0,
+            last_error: None,
+            next_attempt_at: 0,
+        })
+        .unwrap();
+    db.enqueue_op(&PendingOp {
+        id: 0,
+        kind: OP_CREATE.to_string(),
+        uid: "local~child".to_string(),
+        parent_uid: Some(local.to_string()),
+        name: Some("inside.txt".to_string()),
+        blob_path: None,
+        meta_json: None,
+        created_at: 2,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    })
+    .unwrap();
+
+    let newer = db
+        .finish_create(id, Some("/staging/sent"), local, &real, |_| {
+            panic!("an unchanged create has no sidecar to rewrite")
+        })
+        .unwrap();
+
+    assert_eq!(newer, None);
+    let ops = db.pending_ops().unwrap();
+    assert_eq!(ops.len(), 1, "only the child is left");
+    assert_eq!(ops[0].parent_uid.as_deref(), Some(real.as_str()));
+}
+
 #[test]
 fn draining_a_folder_repoints_its_queued_children() {
     let db = Db::open_in_memory().unwrap();