diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index cf735be..0329d87 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -2365,14 +2365,29 @@ impl Core { warn!(%uid, error = %e, "persisting our sealed revision failed"); } } + // A write open from here until the tree has the node takes it as + // its base: the blob it would have copied is let go of below, and + // the tree's stamp names the revision this one replaced + // (docs/BUGS.md B170). + self.landed.lock().insert(uid.clone(), node.clone()); self.rebase_open_writes(uid, &node); } self.release_pending(uid, blob); + #[cfg(test)] + self.drive.landing(uid); + self.place_landed(uid, &node); + self.landed.lock().remove(uid); + Some(node) + } + + /// Put a node whose upload just landed into the tree, unless a newer write + /// is queued for it. + fn place_landed(&self, uid: &NodeUid, node: &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 // this overwrites it with the revision the server actually holds. - if self.rebaseline_pending(uid, &node) { - return Some(node); + if self.rebaseline_pending(uid, node) { + return; } // A rename queued behind the upload has not reached Drive yet, and // until it lands the tree keeps the name and folder the user gave the @@ -2402,7 +2417,6 @@ impl Core { } st.intern(parent, node); }); - Some(node) } /// Rebase every write handle open on `uid` onto `node`, a revision this diff --git a/crates/pdfs-fuse/src/drive.rs b/crates/pdfs-fuse/src/drive.rs index 46e64c0..2fd3bf5 100644 --- a/crates/pdfs-fuse/src/drive.rs +++ b/crates/pdfs-fuse/src/drive.rs @@ -153,6 +153,12 @@ pub(crate) trait DriveApi: Send + Sync { /// The account's storage, across every Proton product. async fn quota(&self) -> Result; + + /// Called by the drain once it has let go of the blob of an upload to + /// `uid` that landed, before the tree has the node, so a simulation can + /// hold it there. Blocks the drain worker it is called on. + #[cfg(test)] + fn landing(&self, _uid: &NodeUid) {} } /// The generic signatures of the client, over the object-safe ones above. diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index 4fa8419..4ba8b89 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -1217,7 +1217,7 @@ impl ProtonFs { /// open. fn serve_open(&self, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) { let write_requested = flags.acc_mode() != OpenAccMode::O_RDONLY; - let (uid, base_mtime, base_size, provisional, base_revision_id) = { + let (uid, mut base_mtime, mut base_size, mut provisional, mut base_revision_id) = { let mut st = self.core.state(); match st.entries.get_mut(&ino.0) { Some(e) if e.node.is_file() => { @@ -1283,22 +1283,24 @@ impl ProtonFs { // `reply` is consumed by whichever arm answers, so the loop yields an // outcome and the answer is sent once, after it. let outcome = 'install: { - // The handle starts at the base's size. A provisional one is the - // ciphertext size, so the file would grow by the encryption overhead - // and reading the base would fail (bugs.md B100). A queued revision - // stamps a real size, so this only goes to the network for a file - // the listing has not sized yet. - let base_size = match provisional && self.core.pending_blob(&uid).is_none() { - false => base_size, - true => match self.core.resolve_provisional_size(ino.0, &uid) { - Some(size) => size, - None => { - error!(%uid, "could not settle the size of the base for a write open"); - break 'install Err(Errno::EIO); - } - }, - }; for attempt in 0..OPEN_BASE_ATTEMPTS { + // The handle starts at the base's size. A provisional one is the + // ciphertext size, so the file would grow by the encryption + // overhead and reading the base would fail (bugs.md B100). A + // queued revision stamps a real size, so this only goes to the + // network for a file the listing has not sized yet. + if provisional && self.core.pending_blob(&uid).is_none() { + match self.core.resolve_provisional_size(ino.0, &uid) { + Some(size) => { + base_size = size; + provisional = false; + } + None => { + error!(%uid, "could not settle the size of the base for a write open"); + break 'install Err(Errno::EIO); + } + } + } let pending_base = self.core.pending.lock().get(&uid).cloned(); if pending_base.as_ref().is_some_and(|p| !p.meta.complete) { error!(%uid, "refusing write over incomplete queued revision"); @@ -1332,17 +1334,49 @@ impl ProtonFs { // a lock-tight comparison โ€” against the multi-gigabyte copy // above, which is the window that actually loses data. let current = self.core.pending_blob(&uid); + let landed = self.core.landed.lock().get(&uid).cloned(); let mut st = self.core.state(); // Only a handle this open is about to *create* can carry a stale // base; if one already exists, that handle is authoritative and // the scratch prepared above is discarded either way. let creating = !st.active_writes.contains_key(&ino.0); - if creating && current != pending_base.map(|p| p.path) { + if creating && current.as_ref() != pending_base.as_ref().map(|p| &p.path) { drop(st); let _ = std::fs::remove_file(&path); warn!(%uid, attempt, "queued revision changed while opening for write; retrying"); continue; } + // With no queued revision, the base's bytes come from the cache, + // keyed by the base's mtime and size. The ones sampled above can + // name a revision this daemon has since replaced: an upload that + // lands lets go of its blob before the tree has its node, and + // the cache holds its bytes under the node only. So the base is + // the landing node while there is one, and the tree's otherwise + // (docs/BUGS.md B170). + if creating && pending_base.is_none() { + let now = landed.or_else(|| st.entries.get(&ino.0).map(|e| e.node.clone())); + if let Some(node) = now + && (node.modification_time != base_mtime + || node_revision_id(&node) != base_revision_id) + { + base_mtime = node.modification_time; + base_size = node_size(&node); + base_revision_id = node_revision_id(&node); + provisional = matches!( + &node.kind, + NodeKind::File { + claimed_size: None, + .. + } + ); + debug!(%uid, base_mtime, base_size, "write open took the base the file has now"); + if provisional { + drop(st); + let _ = std::fs::remove_file(&path); + continue; + } + } + } let fh = st.next_fh; st.next_fh += 1; let aw = st.active_writes.entry(ino.0).or_insert_with(|| { diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 08fe9b2..d6162b2 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -579,6 +579,9 @@ struct Core { /// The SHA-1 of the last revision this daemon uploaded to a file and could /// not read back, so its revision id is not known (`docs/BUGS.md` B157). sealed_unread: Arc>>, + /// The node an upload has just put on Drive, from when the drain lets go + /// of its blob until the tree has the node (`docs/BUGS.md` B170). + landed: Arc>>, /// Creates that landed but could not be read back, by the uid Drive gave /// them, with the placeholder's. The drain adopts one before it sends /// anything else for it. diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 8547e5e..884548b 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -436,6 +436,7 @@ pub(crate) fn mount_with( creating: Arc::new(Mutex::new(HashMap::new())), create_sent: Arc::new(Mutex::new(HashMap::new())), sealed_unread: Arc::new(Mutex::new(HashMap::new())), + landed: Arc::new(Mutex::new(HashMap::new())), unadopted: Arc::new(Mutex::new(HashMap::new())), drain_wake: Arc::new((Mutex::new(false), Condvar::new())), shutdown: Arc::new(crate::shutdown::Shutdown::default()), diff --git a/crates/pdfs-fuse/src/sim/daemon.rs b/crates/pdfs-fuse/src/sim/daemon.rs index 28146cf..a0e1321 100644 --- a/crates/pdfs-fuse/src/sim/daemon.rs +++ b/crates/pdfs-fuse/src/sim/daemon.rs @@ -422,6 +422,47 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + #[test] + #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"] + fn a_write_opened_as_an_upload_lands_takes_the_landed_revision_as_its_base() { + // An upload landed and the drain let go of its blob before the tree had + // the node. A partial write opened in between took the revision the + // upload replaced as its base, found none of its bytes cached, and + // queued an incomplete revision: the next write open failed with EIO + // (B170). + let drive = FakeDrive::new(); + let dir = scratch("write-opened-as-upload-lands"); + let daemon = Daemon::start(&dir, drive.client(1, Faults::lan())).unwrap(); + let path = daemon.mountpoint.join("f.txt"); + let file = |bytes: &[u8]| Some(Entry::File(Arc::new(bytes.to_vec()))); + + std::fs::write(&path, b"first").unwrap(); + assert!(wait_until(Duration::from_secs(30), || { + daemon.pending().is_ok_and(|items| items.is_empty()) + && drive.tree().get("f.txt").cloned() == file(b"first") + })); + let uid = drive.lookup("f.txt").unwrap(); + let landing = daemon.client.hold_next_landing(&uid); + std::fs::write(&path, b"0123456789").unwrap(); + assert!(wait_until(Duration::from_secs(30), || landing.reached())); + write_at(&path, 2, b"ab").unwrap(); + drop(landing); + write_at(&path, 6, b"cd").unwrap(); + assert!( + wait_until(Duration::from_secs(30), || { + daemon.pending().is_ok_and(|items| items.is_empty()) + && drive.tree().get("f.txt").cloned() == file(b"01ab45cd89") + }), + "{:?} {:?}", + daemon.pending(), + drive.tree() + ); + assert_eq!(drive.tree().len(), 1, "{:?}", drive.tree()); + + assert!(matches!(daemon.stop(), Ok(MountOutcome::Shutdown))); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] #[ignore = "mounts FUSE: run with `cargo test -p pdfs-fuse sim:: -- --ignored`"] fn a_file_moved_while_its_create_was_on_the_wire_keeps_its_new_name() { diff --git a/crates/pdfs-fuse/src/sim/fake_drive.rs b/crates/pdfs-fuse/src/sim/fake_drive.rs index 451b48d..eea4812 100644 --- a/crates/pdfs-fuse/src/sim/fake_drive.rs +++ b/crates/pdfs-fuse/src/sim/fake_drive.rs @@ -338,6 +338,7 @@ impl FakeDrive { hold: Mutex::new(None), hold_answer: Mutex::new(None), hold_trash: Mutex::new(None), + hold_landing: Mutex::new(None), drop_after_create: AtomicBool::new(false), lose_create_reply: AtomicBool::new(false), hold_create_reply: Mutex::new(None), @@ -615,6 +616,8 @@ struct ClientInner { hold_answer: Mutex>>, /// The trash [`FakeClient::hold_next_trash`] is waiting for. hold_trash: Mutex>>, + /// The landing [`FakeClient::hold_next_landing`] is waiting for. + hold_landing: Mutex>>, /// Set by [`FakeClient::drop_link_after_next_create`]. drop_after_create: AtomicBool, /// Set by [`FakeClient::lose_reply_to_next_create`]. @@ -752,6 +755,19 @@ impl FakeClient { Held(state) } + /// Hold the drain once the next upload to `uid` has landed and its blob + /// is let go of, before the tree has the node (`DriveApi::landing`), until + /// the answer is dropped. + pub(crate) fn hold_next_landing(&self, uid: &NodeUid) -> Held { + let state = Arc::new(HoldState { + uid: uid.clone(), + reached: AtomicBool::new(false), + released: AtomicBool::new(false), + }); + *self.0.hold_landing.lock() = Some(state.clone()); + Held(state) + } + /// 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 { @@ -1465,6 +1481,21 @@ impl DriveApi for FakeClient { outcome_stream(self.batch(uids, delete_one)) } + fn landing(&self, uid: &NodeUid) { + let state = { + let mut hold = self.0.hold_landing.lock(); + match hold.as_ref() { + Some(state) if state.uid == *uid => hold.take(), + _ => None, + } + }; + let Some(state) = state else { return }; + state.reached.store(true, Ordering::SeqCst); + while !state.released.load(Ordering::SeqCst) { + std::thread::sleep(Duration::from_millis(5)); + } + } + async fn quota(&self) -> Result { self.request(false, |server| { Ok(Quota { diff --git a/docs/BUGS.md b/docs/BUGS.md index 9ae422d..54c4f72 100644 --- a/docs/BUGS.md +++ b/docs/BUGS.md @@ -12,6 +12,37 @@ Conventions: --- +## B170 โ€” A write opened as an upload lands fails the next write with `EIO` + +**Status:** Fixed (unverified). +**Found:** 2026-10-04, by CI on 9bbc994 and by three parallel copies of +`one_client_on_a_good_link` (seed 4; seed 5 in the local suite). A write at an offset was +refused with `EIO`, and the daemon logged "refusing write over incomplete queued revision". +Just before, an open had retried with "queued revision changed while opening for write", and +the queued write it then made was incomplete. + +**Where:** `serve_open` in `crates/pdfs-fuse/src/filesystem.rs`, and `Core::refresh_after_upload` +in `crates/pdfs-fuse/src/drain.rs`. + +**Cause.** When an upload lands, the drain caches its bytes under the node as Drive has it, +rebases the open write handles onto it, lets go of the blob, and then puts the node in the +tree. A write open in between found no blob to copy and took the base it had sampled at its +start: the mtime and size the tree had before the upload. The release of its partial write +filled the gaps from the cache under that key, found nothing, and queued an incomplete +revision, and a write open over one is refused (B123). Under load, the open retried because +of the landing and met that window. + +**Fix.** From before the drain lets go of the blob until the tree has the node, the landing +node is kept aside. A write open with no queued revision takes its base from that node, or +else from the tree as it is when the handle goes in, not from its first sample. + +**Test:** `a_write_opened_as_an_upload_lands_takes_the_landed_revision_as_its_base` in +`sim/daemon.rs` holds the drain in that window with the fake Drive's new `hold_next_landing`, +writes at an offset, and writes again after it. Without the fix the second write fails with +`EIO`. + +--- + ## B169 โ€” A stop lets a mirror pass go on until the runtime ends under it **Status:** Fixed. diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index e1958f2..3cbe2ba 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -68,6 +68,8 @@ by its local id, in the queue, the stored tree and pins alike. to upload them to the top of your Drive instead (B168). - **Stopping the daemon during a folder sync no longer waits for the sync.** The uploads in progress finish, and the rest waits for the next start (B169). +- **Writing to a file again just as its upload finishes no longer fails.** A write right after + it failed with "Input/output error" (B170). - **A file renamed just as its upload finished keeps its new name.** The mount no longer showed the old name, and a second rename no longer failed with "No such file or directory" (B152). - **A change that waits for a name to be freed no longer waits seconds after it is.** A rename or diff --git a/docs/MILESTONE-3.0.0-STATUS.md b/docs/MILESTONE-3.0.0-STATUS.md index 669b289..292e5d4 100644 --- a/docs/MILESTONE-3.0.0-STATUS.md +++ b/docs/MILESTONE-3.0.0-STATUS.md @@ -9,7 +9,7 @@ Tracks `docs/MILESTONE-3.0.0.md`. Update it with every commit that moves a phase - **Phase 2**: local ids, inode = local id, one tree for every mount (2.10.0). - **Phase 3, first part**: mkdir, create, rename, unlink and rmdir on the mount are recorded locally and sent from the queue; `"local_first": false` brings back the old path (5e0f30c). -- Fixes from the simulation and account runs since then: B127 and B141 to B169. +- Fixes from the simulation and account runs since then: B127 and B141 to B170. - **Sync issues, first cut** (ยง6.2): refusals sorted into issues, shown at once in the app, tray and CLI; Export button and `pdfs sync export`; `pdfs sync issues`. @@ -110,7 +110,9 @@ Tracks `docs/MILESTONE-3.0.0.md`. Update it with every commit that moves a phase Looking into it found the start spending 1.7 s rebuilding listings (B167, fixed). Its journal check caught a file removed with its folder while uploading being sent again, to the root (B168, fixed), and a stop that let a mirror pass go on - until the runtime ended under it (B169, fixed). + until the runtime ended under it (B169, fixed). CI on 9bbc994 failed + `one_client_on_a_good_link`: a write opened as an upload landed took the revision + it replaced as its base, and the next write failed with `EIO` (B170, fixed). 5. **Release**: 2.10.0 is tagged; the 2.9.0 tag still waits for the user's go-ahead. ## After 3.0.0