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

@@ -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<NodeUid, Box<dyn std::error::Error>> {
         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<NodeUid, Box<dyn std::error::Error>> {
         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() {

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<T>(
     call: impl Future<Output = Result<T, ProtonError>>,
 ) -> Result<T, ProtonError> {
-    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<T>(
+    limit: Duration,
+    call: impl Future<Output = Result<T, ProtonError>>,
+) -> Result<T, ProtonError> {
+    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<T>(
+        &self,
+        limit: Duration,
+        call: impl Future<Output = Result<T, ProtonError>>,
+    ) -> Result<T, ProtonError> {
+        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);

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

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.