fix(fuse): refuse a move between locations under a just-opened file (B207)

The account run of 3.3.0 failed "move refuses a file still being
written" on the on-demand to mirror pairing: pdfs move moved the folder
while open.txt in it was open for writing, and the write landed after
the move.

retire_create re-keys a landed create's nodes row to the real uid first
and repoints the tree and the open write handles after. A file opened in
between holds a handle with the placeholder uid. has_pending_work_under
placed each open handle by looking its uid up in the nodes table, which
no longer has the placeholder, so it saw nothing open under the folder,
and with the queue already empty the move went ahead. Open handles are
now placed in the in-memory tree by inode, which a landing create does
not change.

The simulation test holds the drain at that point with the new
FakeClient::hold_next_landing_of_any, and fails without the fix.

crates/pdfs-fuse/src/relocate.rs

@@ -25,6 +25,7 @@ use proton_drive_rs::NodeKind;
 use proton_drive_rs::proton_sdk::ids::NodeUid;
 use tracing::{info, warn};
 
+use super::state::State;
 use super::sync::mirror_subtree_unsynced;
 use super::{Core, is_local_uid, parse_uid};
 
@@ -130,6 +131,24 @@ fn rename_noreplace(from: &Path, to: &Path) -> std::io::Result<()> {
     }
 }
 
+/// Whether inode `ino` is `root` or lies under it in the tree. Bounded, so a
+/// parent cycle in corrupt data cannot hang the caller.
+fn ino_is_under(st: &State, mut ino: u64, root: u64) -> bool {
+    for _ in 0..MAX_TREE_DEPTH {
+        if ino == root {
+            return true;
+        }
+        match st.entries.get(&ino) {
+            Some(entry) if entry.parent != ino => ino = entry.parent,
+            _ => return false,
+        }
+    }
+    false
+}
+
+/// How deep [`ino_is_under`] walks before it gives up.
+const MAX_TREE_DEPTH: usize = 1024;
+
 /// Whether `uid` is `root` or lies under it, as far as the nodes table knows.
 fn under(core: &Core, root: &NodeUid, uid: &str) -> bool {
     let root = root.to_string();
@@ -336,7 +355,7 @@ impl Core {
                     core.require_uid_writable(authority)
                         .map_err(|e| core.errno_error(e, "move access"))?;
                 }
-                if self.has_pending_work_under(&uid)? {
+                if self.has_pending_work_under(ino, &uid)? {
                     return Err(CoreError::conflict(
                         "it has changes that are not uploaded yet; wait for the upload to finish",
                     ));
@@ -410,10 +429,11 @@ impl Core {
         Ok(uid)
     }
 
-    /// Whether the queue owes Drive anything for `uid` or a node under it, or a
-    /// file under it is open for writing. Moving such a node away would strand
-    /// the pending change against a path that no longer leads to it.
-    fn has_pending_work_under(&self, uid: &NodeUid) -> CoreResult<bool> {
+    /// Whether the queue owes Drive anything for `uid` (inode `ino`) or a node
+    /// under it, or a file under it is open for writing. Moving such a node away
+    /// would strand the pending change against a path that no longer leads to
+    /// it.
+    fn has_pending_work_under(&self, ino: u64, uid: &NodeUid) -> CoreResult<bool> {
         let ops = self
             .db
             .pending_ops()
@@ -427,12 +447,18 @@ impl Core {
         }) {
             return Ok(true);
         }
-        // Collected first so no State lock is held across the nodes-table reads.
-        let mut writing = Vec::new();
+        // Open handles are placed in the tree by inode, not by uid in the nodes
+        // table. A file opened just after its create landed holds the
+        // placeholder uid until the tree catches up, and the nodes table no
+        // longer has that uid, so the move went ahead under an open write.
+        let mut writing = false;
         self.for_each_state(|st| {
-            writing.extend(st.active_writes.values().map(|w| w.uid.to_string()));
+            writing = st
+                .active_writes
+                .keys()
+                .any(|&open| ino_is_under(st, open, ino));
         });
-        Ok(writing.iter().any(|w| under(self, uid, w)))
+        Ok(writing)
     }
 
     /// Bring the source location level with a move Drive has made.

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

@@ -2185,4 +2185,112 @@ mod tests {
         assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown)));
         let _ = std::fs::remove_dir_all(&dir);
     }
+
+    /// Register `local` as a synced folder in `mode` and wait until the daemon
+    /// lists it that way, returning its id.
+    fn sync_folder(daemon: &Daemon, local: &Path, mode: &str) -> i64 {
+        std::fs::create_dir_all(local).unwrap();
+        let path = local.canonicalize().unwrap().to_string_lossy().into_owned();
+        daemon
+            .request(&Request::AddSyncFolder {
+                local_path: path.clone(),
+            })
+            .unwrap();
+        let listed = |mode: &str| match daemon.request(&Request::ListSyncFolders) {
+            Ok(Response::SyncFolders { items }) => items
+                .into_iter()
+                .find(|f| {
+                    f.local_path == path
+                        && f.mode == mode
+                        && f.pending_mode.is_none()
+                        && f.last_sync > 0
+                })
+                .map(|f| f.id),
+            _ => None,
+        };
+        assert!(
+            wait_until(Duration::from_secs(30), || listed("mirror").is_some()),
+            "{} was registered: {:?}",
+            local.display(),
+            daemon.request(&Request::ListSyncFolders)
+        );
+        let id = listed("mirror").unwrap();
+        // Reads lag behind creates, so the remote folder the first pass made
+        // is not readable at once; switching modes reads it.
+        std::thread::sleep(Duration::from_millis(200));
+        if mode != "mirror" {
+            daemon
+                .request(&Request::SetSyncFolderMode {
+                    id,
+                    mode: mode.into(),
+                })
+                .unwrap();
+            assert!(
+                wait_until(Duration::from_secs(30), || listed(mode).is_some()
+                    && is_mounted(local)),
+                "{} switched to {mode}: {:?}",
+                local.display(),
+                daemon.request(&Request::ListSyncFolders)
+            );
+        }
+        id
+    }
+
+    #[test]
+    #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"]
+    fn a_move_out_of_an_on_demand_folder_refuses_a_file_still_open_for_writing() {
+        // The account run's "move refuses a file still being written" case let
+        // a move from an on-demand synced folder into a mirror folder go ahead
+        // while the file was open, so the write landed after the move.
+        let drive = FakeDrive::new();
+        let dir = scratch("move-open-out-of-ondemand");
+        let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap();
+        let ondemand = dir.join("ondemand");
+        let mirror = dir.join("mirror");
+        sync_folder(&daemon, &ondemand, "ondemand");
+        sync_folder(&daemon, &mirror, "mirror");
+
+        let source = ondemand.join("open");
+        std::fs::create_dir(&source).unwrap();
+        daemon.request(&Request::SyncNow { id: None }).unwrap();
+        assert!(wait_until(Duration::from_secs(30), || {
+            daemon.pending().is_ok_and(|items| items.is_empty())
+        }));
+        // The account run opened the file a moment after its create landed:
+        // the queue was already empty, and the tree still had the
+        // placeholder uid it gave the file.
+        let landing = daemon.client.hold_next_landing_of_any();
+        {
+            let mut file = std::fs::File::create(source.join("open.txt")).unwrap();
+            std::io::Write::write_all(&mut file, b"before").unwrap();
+            file.sync_all().unwrap();
+        }
+        assert!(wait_until(Duration::from_secs(30), || landing.reached()));
+
+        let mut file = std::fs::OpenOptions::new()
+            .append(true)
+            .open(source.join("open.txt"))
+            .unwrap();
+        std::io::Write::write_all(&mut file, b" and after").unwrap();
+        let moved = daemon.request(&Request::Move {
+            path: source.to_string_lossy().into_owned(),
+            new_parent: mirror
+                .canonicalize()
+                .unwrap()
+                .to_string_lossy()
+                .into_owned(),
+        });
+        drop(file);
+        drop(landing);
+        match moved {
+            Err(e) => assert!(
+                e.to_string().contains("not uploaded yet"),
+                "refused for another reason: {e}"
+            ),
+            Ok(answer) => panic!("the move went ahead: {answer:?}"),
+        }
+
+        assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown)));
+        let _ = std::fs::remove_dir_all(&dir);
+    }
 }

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

@@ -362,6 +362,7 @@ impl FakeDrive {
             hold: Mutex::new(None),
             hold_answer: Mutex::new(None),
             hold_trash: Mutex::new(None),
+            hold_any_landing: AtomicBool::new(false),
             hold_landing: Mutex::new(None),
             drop_after_create: AtomicBool::new(false),
             lose_create_reply: AtomicBool::new(false),
@@ -642,6 +643,9 @@ struct ClientInner {
     hold_trash: Mutex<Option<Arc<HoldState>>>,
     /// The landing [`FakeClient::hold_next_landing`] is waiting for.
     hold_landing: Mutex<Option<Arc<HoldState>>>,
+    /// Set by [`FakeClient::hold_next_landing_of_any`]: hold whichever node
+    /// lands next.
+    hold_any_landing: AtomicBool,
     /// Set by [`FakeClient::drop_link_after_next_create`].
     drop_after_create: AtomicBool,
     /// Set by [`FakeClient::lose_reply_to_next_create`].
@@ -792,6 +796,14 @@ impl FakeClient {
         Held(state)
     }
 
+    /// [`FakeClient::hold_next_landing`] for whichever node lands next, such
+    /// as a new file, whose uid Drive has not minted yet.
+    pub(crate) fn hold_next_landing_of_any(&self) -> Held {
+        let held = self.hold_next_landing(&uid("*"));
+        self.0.hold_any_landing.store(true, Ordering::SeqCst);
+        held
+    }
+
     /// Make the next file create on Drive, then hold its answer until the
     /// answer is dropped, as a slow link does.
     pub(crate) fn hold_reply_to_next_create(&self) -> Held {
@@ -1508,8 +1520,12 @@ impl DriveApi for FakeClient {
     fn landing(&self, uid: &NodeUid) {
         let state = {
             let mut hold = self.0.hold_landing.lock();
+            let any = self.0.hold_any_landing.load(Ordering::SeqCst);
             match hold.as_ref() {
-                Some(state) if state.uid == *uid => hold.take(),
+                Some(state) if any || state.uid == *uid => {
+                    self.0.hold_any_landing.store(false, Ordering::SeqCst);
+                    hold.take()
+                }
                 _ => None,
             }
         };