fix(fuse): do not send a file removed with its folder again (B168)

A delete of a file whose create was on the wire withdraws the create into
a trash. When its folder was removed too and that trash landed first,
Drive refused the create, and the drain re-homed the deleted file in the
root; only its missing bytes stopped it. The withdrawn create's trash
then read the trashed folder back into the cache before finding nothing
to trash.

A withdrawn create is no longer re-homed, and its trash is done once the
folder it was sent to is trashed or gone.

crates/pdfs-fuse/src/drain.rs

@@ -534,7 +534,32 @@ impl Core {
     ///   and `pdfs recover` is how they get them back.
     /// * we could not ask. Ordinary deferral; the network is the network.
     fn resolve_unknown_authority(&self, op: &PendingOp, authority: &NodeUid) {
-        match self.fetch_node_remote(authority) {
+        let fetched = self.fetch_node_remote(authority);
+        // The trash a withdrawn create left looks for the node it may have made
+        // in the folder it was sent to. A folder trashed or gone took that node
+        // with it, so there is nothing left to trash. Most often it is gone
+        // because the user removed it with the file, and its trash landed first
+        // (`docs/BUGS.md` B168).
+        let folder_gone = match &fetched {
+            Ok(None) => true,
+            Ok(Some(node)) => node.trashed,
+            Err(_) => false,
+        };
+        if op.kind == OP_TRASH
+            && is_local_uid_str(&op.uid)
+            && folder_gone
+            && let Some(uid) = parse_node_uid(&op.uid)
+        {
+            debug!(uid = %op.uid, %authority, "a withdrawn create's folder is gone; trash op satisfied");
+            if let Err(error) = self.retire_trash_op(op, &uid) {
+                error!(uid = %op.uid, %error, "retiring a withdrawn create's trash failed");
+                self.release_claim(op);
+                return;
+            }
+            self.create_sent.lock().remove(&op.id);
+            return;
+        }
+        match fetched {
             Ok(Some(node)) => {
                 if let Err(error) = self.db.upsert_node(&node) {
                     error!(uid = %op.uid, %authority, %error,
@@ -1139,8 +1164,13 @@ impl Core {
         // succeed as written and retrying it forever wedges the queue behind a
         // file that has bytes to save. Re-home it to the root: not where the
         // user put it, but it exists, it is visible, and the bytes are intact.
+        //
+        // Not for a create a delete withdrew while it was on the wire: the
+        // folder is most often gone because the user removed it, with this
+        // file, and the file would come back in the root (`docs/BUGS.md` B168).
         if let Some(root) = self.root_uid()
             && real.is_err()
+            && !self.db.has_pending_op(&op.uid, OP_TRASH)?
             && self.parent_is_gone(&parent)
         {
             warn!(%local, name, "parent folder is gone remotely; creating in the root instead");

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

@@ -649,6 +649,48 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    #[test]
+    #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
+    fn a_file_removed_with_its_folder_while_uploading_stays_removed() {
+        // A folder removed while a file in it was on its way to Drive: the
+        // folder's trash landed first, Drive refused the file's create, and
+        // the drain took the folder for trashed elsewhere and made the deleted
+        // file again in the root, failing only because its bytes were gone.
+        // Its queued trash then re-read the folder it had been sent to (B168).
+        let drive = FakeDrive::new();
+        let dir = scratch("removed-while-uploading");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let mnt = daemon.mountpoint.clone();
+        std::fs::create_dir(mnt.join("x")).unwrap();
+        assert!(wait_until(Duration::from_secs(30), || {
+            daemon.pending().is_ok_and(|items| items.is_empty()) && drive.lookup("x").is_some()
+        }));
+        let folder = drive.lookup("x").unwrap();
+
+        let create = daemon.client.hold_next_create();
+        std::fs::write(mnt.join("x/removed-while-uploading"), b"abc").unwrap();
+        assert!(wait_until(Duration::from_secs(30), || create.reached()));
+        std::fs::remove_file(mnt.join("x/removed-while-uploading")).unwrap();
+        std::fs::remove_dir(mnt.join("x")).unwrap();
+        assert!(wait_until(Duration::from_secs(30), || drive
+            .lookup("x")
+            .is_none()));
+        drop(create);
+        assert!(wait_until(Duration::from_secs(60), || {
+            daemon.pending().is_ok_and(|items| items.is_empty())
+        }));
+
+        assert_eq!(drive.names_ever_in_root(), ["x"]);
+        assert!(drive.tree().is_empty());
+        let rehomed = take_logged().into_iter().find(|line| {
+            line.contains("creating in the root") && line.contains("removed-while-uploading")
+        });
+        assert_eq!(rehomed, None);
+        assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown)));
+        let db = Db::open(&dir.join("pdfs.db")).unwrap();
+        assert!(db.node_by_uid(&folder.to_string()).unwrap().is_none());
+    }
+
     #[test]
     #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
     fn a_folder_made_again_while_the_old_one_is_trashed_is_a_new_folder() {

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

@@ -38,7 +38,8 @@
 //! also have a new file's next reads answered as Drive lists it before its
 //! create is committed ([`FakeClient::answer_reads_unrevised`]; B163), and
 //! hold a folder's listing on its way back with one of its files listed that
-//! way ([`FakeClient::hold_answer_to_next_listing`]).
+//! way ([`FakeClient::hold_answer_to_next_listing`]), or hold a new file's
+//! create on its way to Drive ([`FakeClient::hold_next_create`]).
 
 use std::collections::{BTreeMap, HashMap};
 use std::io::{Read, Write};
@@ -340,6 +341,7 @@ impl FakeDrive {
             drop_after_create: AtomicBool::new(false),
             lose_create_reply: AtomicBool::new(false),
             hold_create_reply: Mutex::new(None),
+            hold_create: Mutex::new(None),
             lose_revision_reply: AtomicBool::new(false),
             lose_revision_read_back: AtomicBool::new(false),
             lose_read: Mutex::new(None),
@@ -405,6 +407,23 @@ impl FakeDrive {
         tree
     }
 
+    /// Every name a node has had in the root, trashed or not, as far back as
+    /// Drive keeps its versions.
+    pub(crate) fn names_ever_in_root(&self) -> Vec<String> {
+        let server = self.server.lock();
+        let mut names: Vec<String> = server
+            .nodes
+            .values()
+            .flatten()
+            .filter_map(|(_, remote)| remote.as_ref())
+            .filter(|remote| remote.parent.as_ref() == Some(&server.root))
+            .map(|remote| remote.name.clone())
+            .collect();
+        names.sort();
+        names.dedup();
+        names
+    }
+
     /// The uid at `path` below the root, as Drive has it now.
     pub(crate) fn lookup(&self, path: &str) -> Option<NodeUid> {
         let server = self.server.lock();
@@ -602,6 +621,8 @@ struct ClientInner {
     lose_create_reply: AtomicBool,
     /// The create [`FakeClient::hold_reply_to_next_create`] is waiting for.
     hold_create_reply: Mutex<Option<Arc<HoldState>>>,
+    /// The create [`FakeClient::hold_next_create`] is waiting for.
+    hold_create: Mutex<Option<Arc<HoldState>>>,
     /// Set by [`FakeClient::lose_reply_to_next_revision`].
     lose_revision_reply: AtomicBool,
     /// Set by [`FakeClient::lose_read_back_of_next_revision`].
@@ -743,6 +764,18 @@ impl FakeClient {
         Held(state)
     }
 
+    /// Hold the next file create before it reaches Drive, until the answer
+    /// is dropped.
+    pub(crate) fn hold_next_create(&self) -> Held {
+        let state = Arc::new(HoldState {
+            uid: self.0.drive.root(),
+            reached: AtomicBool::new(false),
+            released: AtomicBool::new(false),
+        });
+        *self.0.hold_create.lock() = Some(state.clone());
+        Held(state)
+    }
+
     /// Make the next revision of a file on Drive and lose its answer.
     pub(crate) fn lose_reply_to_next_revision(&self) {
         self.0.lose_revision_reply.store(true, Ordering::SeqCst);
@@ -1225,6 +1258,10 @@ impl DriveApi for FakeClient {
         _aead: bool,
     ) -> Result<NodeUid> {
         let content = read_all(reader)?;
+        let held = self.0.hold_create.lock().take();
+        if let Some(state) = held {
+            wait_until_released(&state).await;
+        }
         let made = self
             .request(true, |server| {
                 server.new_file(