From 9b3651e4e88a0e0ea8849bee9a529efccefe03e7 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Sun, 4 Oct 2026 03:42:44 +0200 Subject: [PATCH] fix(fuse): put a deadline on every remote call the drain makes create_folder, trash_nodes, the twin listing and the uploads waited on Drive without a limit, so one call that never answered held a drain thread for good. They now give up after INTERACTIVE_CALL_TIMEOUT, and uploads after that plus their size at 32 KiB/s; a missed deadline reads as RequestTimeout, so the op backs off like any lost link. --- crates/pdfs-fuse/src/drain.rs | 81 +++++++++++++++++----------------- crates/pdfs-fuse/src/link.rs | 53 +++++++++++++++++++++- docs/CHANGELOG.md | 3 ++ docs/MILESTONE-3.0.0-STATUS.md | 3 +- 4 files changed, 97 insertions(+), 43 deletions(-) diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 03db5ed..afdd76c 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -40,7 +40,7 @@ use proton_drive_rs::proton_sdk::ids::NodeUid; use proton_drive_rs::{Node, NodeKind}; use tracing::{debug, error, info, warn}; -use super::link::is_network_error; +use super::link::{self, is_network_error}; use super::state::{Intervals, PendingRevision}; use super::transfers::CountingReader; use super::{ @@ -887,8 +887,7 @@ impl Core { return Ok(()); } match self - .rt - .block_on(self.drive.trash_nodes(std::slice::from_ref(&uid))) + .block_on_bounded(self.drive.trash_nodes(std::slice::from_ref(&uid))) .and_then(batch::into_unit) { Ok(()) => {} @@ -1192,10 +1191,8 @@ impl Core { if racing { return Ok(None); } - let uids = self - .rt - .block_on(self.drive.enumerate_folder_children_node_uids(parent))?; - let children = self.rt.block_on(self.drive.enumerate_nodes_light(&uids))?; + let uids = self.block_on_bounded(self.drive.enumerate_folder_children_node_uids(parent))?; + let children = self.block_on_bounded(self.drive.enumerate_nodes_light(&uids))?; let Some(twin) = children .into_iter() .find(|node| node.name == name && !node.trashed) @@ -1232,11 +1229,11 @@ impl Core { name: &str, ) -> Result> { match op.kind == OP_MKDIR { - true => { - Ok(self - .rt - .block_on(self.drive.create_folder(parent, name, Some(now_secs())))?) - } + true => Ok(self.block_on_bounded(self.drive.create_folder( + parent, + name, + Some(now_secs()), + ))?), false => self.upload_created_file(op, parent, name), } } @@ -1250,7 +1247,7 @@ impl Core { name: &str, ) -> Result> { let Some(blob) = op.blob_path.as_deref() else { - return Ok(self.rt.block_on(self.drive.upload_file( + return Ok(self.block_on_bounded(self.drive.upload_file( parent, name, media_type_for(name), @@ -1270,16 +1267,19 @@ impl Core { .transfers .begin(name, op.uid.clone(), TransferDirection::Upload, meta.len); let reader = CountingReader::new(File::open(blob)?, &guard); - let uid = self.rt.block_on(self.drive.upload_file_from( - parent, - name, - media_type_for(name), - reader, - meta.len as i64, - thumbnails, - None, - false, - ))?; + let uid = self.block_on_within( + link::upload_deadline(meta.len), + self.drive.upload_file_from( + parent, + name, + media_type_for(name), + reader, + meta.len as i64, + thumbnails, + None, + false, + ), + )?; Ok(uid) } @@ -1619,16 +1619,19 @@ impl Core { self.transfers .begin(&alt, meta.uid.clone(), TransferDirection::Upload, meta.len); let reader = CountingReader::new(File::open(blob)?, &guard); - self.rt.block_on(self.drive.upload_file_from( - &parent, - &alt, - media_type_for(&alt), - reader, - meta.len as i64, - thumbnails, - None, - false, - ))?; + self.block_on_within( + link::upload_deadline(meta.len), + self.drive.upload_file_from( + &parent, + &alt, + media_type_for(&alt), + reader, + meta.len as i64, + thumbnails, + None, + false, + ), + )?; drop(guard); self.db.delete_op(op.id)?; @@ -1928,13 +1931,11 @@ impl Core { } let reader = CountingReader::new(File::open(&blob)?, &guard).with_cancel(cancel.flag()); let started = Instant::now(); - let sent = self.rt.block_on(self.drive.upload_new_revision_from( - &uid, - reader, - meta.len as i64, - thumbnails, - None, - )); + let sent = self.block_on_within( + link::upload_deadline(meta.len), + self.drive + .upload_new_revision_from(&uid, reader, meta.len as i64, thumbnails, None), + ); let took = started.elapsed(); if let Err(e) = sent { if cancel.cancelled() { diff --git a/crates/pdfs-fuse/src/link.rs b/crates/pdfs-fuse/src/link.rs index 884ba56..e5dd685 100644 --- a/crates/pdfs-fuse/src/link.rs +++ b/crates/pdfs-fuse/src/link.rs @@ -86,13 +86,32 @@ pub(crate) fn no_answer(after: Duration) -> ProtonError { }) } +/// The slowest upload rate an upload's deadline allows for, past the +/// metadata budget it starts with. Far below any link that works, so only a +/// transfer that has stopped moving runs into it. +const SLOWEST_UPLOAD: u64 = 32 * 1024; + +/// How long an upload of `len` bytes may take before the link is treated as +/// down: the budget of a metadata call, plus the bytes at [`SLOWEST_UPLOAD`]. +pub(crate) fn upload_deadline(len: u64) -> Duration { + INTERACTIVE_CALL_TIMEOUT + Duration::from_secs(len / SLOWEST_UPLOAD) +} + /// Run a remote call, giving up after [`INTERACTIVE_CALL_TIMEOUT`]. pub(crate) async fn bounded( call: impl Future>, ) -> Result { - match tokio::time::timeout(INTERACTIVE_CALL_TIMEOUT, call).await { + bounded_by(INTERACTIVE_CALL_TIMEOUT, call).await +} + +/// Run a remote call, giving up after `limit`. +pub(crate) async fn bounded_by( + limit: Duration, + call: impl Future>, +) -> Result { + match tokio::time::timeout(limit, call).await { Ok(result) => result, - Err(_) => Err(no_answer(INTERACTIVE_CALL_TIMEOUT)), + Err(_) => Err(no_answer(limit)), } } @@ -132,6 +151,16 @@ impl Core { self.rt.block_on(bounded(call)) } + /// Run a remote call, giving up after `limit`: an upload, whose budget + /// grows with its size ([`upload_deadline`]). + pub(crate) fn block_on_within( + &self, + limit: Duration, + call: impl Future>, + ) -> Result { + self.rt.block_on(bounded_by(limit, call)) + } + /// Report a failed remote call. Returns whether it failed for want of a /// network, in which case the mount is now offline and the caller should /// take its queued path instead of failing. @@ -310,6 +339,26 @@ mod tests { assert!(!is_network_error(&ProtonError::invalid_operation("bug"))); } + #[test] + fn an_upload_gets_longer_the_bigger_it_is() { + assert_eq!(upload_deadline(0), INTERACTIVE_CALL_TIMEOUT); + assert_eq!( + upload_deadline(1 << 30), + INTERACTIVE_CALL_TIMEOUT + Duration::from_secs(32_768) + ); + } + + #[test] + fn a_call_past_its_deadline_reads_as_a_lost_link() { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .unwrap(); + let limit = Duration::from_millis(20); + let result: Result<(), _> = rt.block_on(bounded_by(limit, std::future::pending())); + assert!(result.is_err_and(|e| is_network_error(&e))); + } + #[test] fn a_young_outage_is_probed_often_and_an_old_one_rarely() { let young = Duration::from_secs(60); diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 8a0989d..0ca936e 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -31,6 +31,9 @@ are kept. Queued changes get a column for the reason Drive refused them. for this release. ### Fixed +- **A queued change no longer holds up the queue forever on a link that stops answering.** + Sending a new folder, a delete or a file waits at most 20 seconds for Drive, plus time for + the file's size, and then the queue treats the link as down and tries again. - **A file deleted while its first upload is on the wire no longer stays on Drive.** The node the upload made is trashed as soon as it lands, so its name is free again (B129). - **A delete queued while a rename is sent is no longer lost.** The rename landing could remove diff --git a/docs/MILESTONE-3.0.0-STATUS.md b/docs/MILESTONE-3.0.0-STATUS.md index a755b35..1f6d92f 100644 --- a/docs/MILESTONE-3.0.0-STATUS.md +++ b/docs/MILESTONE-3.0.0-STATUS.md @@ -50,7 +50,8 @@ Tracks `docs/MILESTONE-3.0.0.md`. Update it with every commit that moves a phase 3.0.0). - **Phase 4**: one applier for remote changes. - **Phase 5**: mirror folders on the same planner and executor. -- **Any time**: `Size` and block geometry as types; a `Deadline` on every remote call. +- **Any time**: `Size` and block geometry as types; a deadline on the remote calls outside + the drain (the drain's have one: `link::upload_deadline`, `Core::block_on_within`). - Open questions in ยง12 of the milestone. - `PLAN.md`: the acceptance script notices network drops and reports `interrupted` instead of TIMEOUT.