fix(fuse): keep the revision id of a file created through the mount

The write handle a create opens was given no revision id, so the drain's
conflict check fell back to (mtime, size). Drive advances a node's mtime
on a move or rename, so a new file moved before its first upload drained
landed as a (sync-conflict) copy next to its own empty original. Take the
minted node's revision id, and rebase handles still open on a queued
create onto the node it landed as (B113).

crates/pdfs-fuse/src/drain.rs

@@ -1151,6 +1151,12 @@ impl Core {
         if let Err(e) = self.db.upsert_node(&node) {
             warn!(%real, error = %e, "db upsert_node failed after remap");
         }
+        // A handle still open on the placeholder was based on no revision at
+        // all: a local-clock mtime and no revision id. Released as it is, its
+        // write would land as a conflict copy of the empty file this create just
+        // made (`docs/BUGS.md` B113). Rebased before the restamp below, so a
+        // release that races this is caught there.
+        self.rebase_open_writes(real, &node);
         // A write carried over from the create was made against no revision at
         // all; the one just created is what it now replaces.
         self.rebaseline_pending(real, &node);
@@ -1707,14 +1713,11 @@ impl Core {
         // This is the open-handle counterpart of `rebaseline_pending`, which
         // covers the same gap for *queued* ops.
         {
-            let sealed_mtime = node.modification_time;
-            let sealed_size = node_size(&node);
-            let sealed_rev = node_revision_id(&node);
             // Remember it as ours, so a queued write that opened over an earlier
             // revision recognises this as a self-supersede rather than a foreign
             // change and does not fork (B70 layer B; consulted in
             // `revision_conflict`).
-            if let Some(rev) = sealed_rev.clone() {
+            if let Some(rev) = node_revision_id(&node) {
                 self.own_sealed_revs.lock().insert(uid.clone(), rev.clone());
                 // Write through: losing this at shutdown reopened the fork
                 // window B70 layer B closed, for the first drain after a
@@ -1726,17 +1729,7 @@ impl Core {
                     warn!(%uid, error = %e, "persisting our sealed revision failed");
                 }
             }
-            self.for_each_state(|st| {
-                for aw in st.active_writes.values_mut() {
-                    if aw.uid == *uid {
-                        aw.base_mtime = sealed_mtime;
-                        aw.base_size = sealed_size;
-                        aw.base_revision_id = sealed_rev.clone();
-                        debug!(%uid, mtime = sealed_mtime, size = sealed_size,
-                               "rebased open write handle onto the revision just uploaded");
-                    }
-                }
-            });
+            self.rebase_open_writes(uid, &node);
         }
         // Ordered so that a write queued *during* the fetch above is still
         // caught: it took its baseline from the node's optimistic stamp, and
@@ -1757,6 +1750,20 @@ impl Core {
         });
     }
 
+    /// Rebase every write handle open on `uid` onto `node`, a revision this
+    /// daemon has just put there itself (`WriteHandle::rebase_onto`).
+    fn rebase_open_writes(&self, uid: &NodeUid, node: &Node) {
+        self.for_each_state(|st| {
+            for aw in st.active_writes.values_mut() {
+                if aw.uid == *uid {
+                    aw.rebase_onto(node);
+                    debug!(%uid, mtime = aw.base_mtime, size = aw.base_size,
+                           "rebased open write handle onto the revision just landed");
+                }
+            }
+        });
+    }
+
     /// Point a still-queued write at the revision this upload just sealed, and
     /// report whether there was one.
     ///
@@ -1891,6 +1898,7 @@ fn adoptable(is_dir: bool, twin: &Node, op_created_ms: i64) -> bool {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::state::WriteHandle;
     use pdfs_core::Access;
     use pdfs_core::db::Db;
     use proton_drive_rs::NodeKind;
@@ -2317,6 +2325,26 @@ mod tests {
         assert!(revision_changed(&base, &remote).is_some());
     }
 
+    /// The reported shape (docs/BUGS.md B113): a downloader creates a file
+    /// through the mount, and the file is moved before its write drains. The
+    /// move re-stamps the node's mtime on the server (1790955809 -> 1790956593),
+    /// but the revision is still the empty one the create minted.
+    #[test]
+    fn a_move_before_the_first_write_drains_is_not_a_conflict() {
+        let minted = file_node(1_790_955_809, 0, Some("empty"));
+        let path = std::env::temp_dir().join(format!("pdfs-drain-test-{}", std::process::id()));
+        let file = File::create(&path).unwrap();
+        let _ = std::fs::remove_file(&path);
+        let h = WriteHandle::created(2, &minted, file, path);
+        let base = baseline(h.base_mtime, h.base_size, h.base_revision_id.as_deref());
+
+        let moved = file_node(1_790_956_593, 0, Some("empty"));
+        assert_eq!(revision_changed(&base, &moved), None);
+        // Another device writing it is still a conflict.
+        let edited = file_node(1_790_956_593, 0, Some("theirs"));
+        assert!(revision_changed(&base, &edited).is_some());
+    }
+
     #[test]
     fn without_ids_it_falls_back_to_mtime_and_size() {
         // Old sidecar (no revision id): the (mtime, size) tuple still governs.

crates/pdfs-fuse/src/filesystem.rs

@@ -1550,26 +1550,16 @@ impl ProtonFs {
                 }
             },
         };
-        let new_uid = node.uid.clone();
-        // The base this handle writes over is the node as it actually exists —
-        // the empty file the server just minted — so its modification time comes
-        // from the node, never from the local clock. `queue_revision` turns this
-        // into `StagedWrite::based_on` and the drain compares that against the
-        // remote: a `now_secs()` here differs from the server's stamp by however
-        // long the create round-trip took, so the first write to a brand-new file
-        // conflicts with its own create whenever the second happens to tick in
-        // between.
-        let base_mtime = node.modification_time;
         let (file, path) = match self.core.cache.create_scratch() {
             Ok(x) => x,
             Err(e) => {
-                error!(%new_uid, error = %e, "create scratch file failed");
+                error!(uid = %node.uid, error = %e, "create scratch file failed");
                 reply.error(Errno::EIO);
                 return;
             }
         };
         let mut st = self.core.state();
-        let ino = st.intern(parent, node);
+        let ino = st.intern(parent, node.clone());
         if let Some(entry) = st.entries.get_mut(&ino) {
             entry.open_count = entry.open_count.saturating_add(1);
         }
@@ -1580,23 +1570,10 @@ impl ProtonFs {
         }
         let fh = st.next_fh;
         st.next_fh += 1;
-        let aw = st.active_writes.entry(ino).or_insert_with(|| {
-            // A brand-new file: empty base, everything written is authored.
-            WriteHandle {
-                ino,
-                uid: new_uid,
-                file: Arc::new(file),
-                path,
-                written: Intervals::default(),
-                len: 0,
-                base_size: 0,
-                base_mtime,
-                // A brand-new file has no sealed remote revision to conflict with.
-                base_revision_id: None,
-                dirty: false,
-                open_count: 0,
-            }
-        });
+        let aw = st
+            .active_writes
+            .entry(ino)
+            .or_insert_with(|| WriteHandle::created(ino, &node, file, path));
         aw.open_count += 1;
         st.handles.insert(fh, ino);
         let entry = st.entries.get(&ino).unwrap();

crates/pdfs-fuse/src/state.rs

@@ -129,6 +129,44 @@ pub(crate) struct WriteHandle {
 }
 
 impl WriteHandle {
+    /// A handle on a file `create` just made: an empty base, so everything
+    /// written through it is authored.
+    ///
+    /// The base is `node` as it actually exists — the empty file the server just
+    /// minted, or a queued create's placeholder — never the local clock or a
+    /// blank. `queue_revision` turns it into `StagedWrite::based_on` and the
+    /// drain compares the remote against that. A `now_secs()` mtime differs from
+    /// the server's stamp by the create round-trip. A missing revision id drops
+    /// the check to `(mtime, size)`, and moving or renaming the file advances
+    /// its mtime on the server. Either way the first write to a brand-new file
+    /// lands as a conflict copy of its own empty create (`docs/BUGS.md` B113).
+    pub(crate) fn created(ino: u64, node: &Node, file: File, path: PathBuf) -> Self {
+        Self {
+            ino,
+            uid: node.uid.clone(),
+            file: Arc::new(file),
+            path,
+            written: Intervals::default(),
+            len: 0,
+            base_size: 0,
+            base_mtime: node.modification_time,
+            base_revision_id: crate::node_revision_id(node),
+            dirty: false,
+            open_count: 0,
+        }
+    }
+
+    /// Write over `node` from now on: a revision this daemon has just put on the
+    /// server under the handle, by sealing an upload or by landing the queued
+    /// create the handle was opened on. The base the handle opened over is gone,
+    /// and keeping it would make the drain read our own revision as another
+    /// device's edit.
+    pub(crate) fn rebase_onto(&mut self, node: &Node) {
+        self.base_mtime = node.modification_time;
+        self.base_size = crate::node_size(node);
+        self.base_revision_id = crate::node_revision_id(node);
+    }
+
     /// Account for `[offset, end)` having been written through this handle.
     ///
     /// A write past the end leaves a hole that reads as zeros, exactly like a
@@ -1918,6 +1956,27 @@ mod tests {
         }
     }
 
+    /// A file created offline is written while its queued create lands. The
+    /// placeholder had no revision, so the handle takes the empty file the
+    /// create made as its base (docs/BUGS.md B113).
+    #[test]
+    fn a_handle_on_a_placeholder_is_based_on_the_create_that_landed() {
+        let mut h = handle(0, 0);
+        assert_eq!(h.base_revision_id, None);
+        let mut landed = node("real", "root", "f.mkv", false);
+        landed.modification_time = 160;
+        if let NodeKind::File {
+            active_revision_id, ..
+        } = &mut landed.kind
+        {
+            *active_revision_id = Some("empty".into());
+        }
+        h.rebase_onto(&landed);
+        assert_eq!(h.base_mtime, 160);
+        assert_eq!(h.base_size, 0);
+        assert_eq!(h.base_revision_id.as_deref(), Some("empty"));
+    }
+
     #[test]
     fn a_listing_does_not_shrink_a_file_open_for_writing() {
         let (mut st, _dir) = state();