fix(fuse): give uploads under a low cap the time their share needs

The upload deadline assumed at least 32 KiB/s. Under an upload cap shared
by every drain worker and sync task, a large file ran out of time on every
attempt, took the mount offline and restarted from byte 0. The deadline
now counts the upload's share of the cap as the slowest rate when that
share is lower (B231).

crates/pdfs-fuse/src/drain.rs

@@ -1604,7 +1604,7 @@ impl Core {
             .begin(name, op.uid.clone(), TransferDirection::Upload, meta.len);
         let reader = CountingReader::new(File::open(blob)?, &guard);
         let uid = self.block_on_within(
-            link::upload_deadline(meta.len),
+            link::upload_deadline(meta.len, self.transfers.upload_cap()),
             self.drive.upload_file_from(
                 parent,
                 name,
@@ -2001,7 +2001,7 @@ impl Core {
                 .begin(&alt, meta.uid.clone(), TransferDirection::Upload, meta.len);
         let reader = CountingReader::new(File::open(blob)?, &guard);
         self.block_on_within(
-            link::upload_deadline(meta.len),
+            link::upload_deadline(meta.len, self.transfers.upload_cap()),
             self.drive.upload_file_from(
                 &parent,
                 &alt,
@@ -2329,7 +2329,7 @@ impl Core {
         let reader = CountingReader::new(File::open(&blob)?, &guard).with_cancel(cancel.flag());
         let started = Instant::now();
         let sent = self.block_on_within(
-            link::upload_deadline(meta.len),
+            link::upload_deadline(meta.len, self.transfers.upload_cap()),
             self.drive
                 .upload_new_revision_from(&uid, reader, meta.len as i64, thumbnails, None),
         );

crates/pdfs-fuse/src/link.rs

@@ -119,10 +119,25 @@ pub(crate) fn no_answer(after: Duration) -> ProtonError {
 /// transfer that has stopped moving runs into it.
 const SLOWEST_UPLOAD: u64 = 32 * 1024;
 
+/// How many uploads can share the upload cap at once: one per drain worker
+/// and one per mirror sync task.
+const UPLOAD_SLOTS: u64 = (crate::DRAIN_WORKERS + crate::sync::SYNC_CONCURRENCY) as u64;
+
 /// 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)
+///
+/// An upload cap of `cap` bytes per second (`0` for none) is shared by every
+/// upload running, so each may get as little as its share of it. Below
+/// [`SLOWEST_UPLOAD`] a share counts as the slowest rate instead: measured
+/// against the fixed rate, a large file under a low cap ran out of time on
+/// every attempt, took the mount offline each time, and started over from
+/// its first byte.
+pub(crate) fn upload_deadline(len: u64, cap: u64) -> Duration {
+    let slowest = match cap {
+        0 => SLOWEST_UPLOAD,
+        cap => SLOWEST_UPLOAD.min((cap / UPLOAD_SLOTS).max(1)),
+    };
+    INTERACTIVE_CALL_TIMEOUT + Duration::from_secs(len / slowest)
 }
 
 /// Run a remote call, giving up after [`INTERACTIVE_CALL_TIMEOUT`].
@@ -410,13 +425,31 @@ mod tests {
 
     #[test]
     fn an_upload_gets_longer_the_bigger_it_is() {
-        assert_eq!(upload_deadline(0), INTERACTIVE_CALL_TIMEOUT);
+        assert_eq!(upload_deadline(0, 0), INTERACTIVE_CALL_TIMEOUT);
         assert_eq!(
-            upload_deadline(1 << 30),
+            upload_deadline(1 << 30, 0),
             INTERACTIVE_CALL_TIMEOUT + Duration::from_secs(32_768)
         );
     }
 
+    #[test]
+    fn an_upload_under_a_low_cap_gets_the_time_its_share_of_the_cap_needs() {
+        // A generous cap leaves the deadline as it was.
+        assert_eq!(
+            upload_deadline(1 << 30, 100 << 20),
+            upload_deadline(1 << 30, 0)
+        );
+        // 512 KiB/s shared by every upload slot is below the slowest rate.
+        let share = (512 << 10) / UPLOAD_SLOTS;
+        assert!(share < SLOWEST_UPLOAD);
+        assert_eq!(
+            upload_deadline(1 << 30, 512 << 10),
+            INTERACTIVE_CALL_TIMEOUT + Duration::from_secs((1 << 30) / share)
+        );
+        // A cap below one byte per slot still gives a finite deadline.
+        assert!(upload_deadline(1 << 20, 1) > upload_deadline(1 << 20, 512 << 10));
+    }
+
     #[test]
     fn a_call_past_its_deadline_reads_as_a_lost_link() {
         let rt = tokio::runtime::Builder::new_current_thread()

crates/pdfs-fuse/src/sync.rs

@@ -47,7 +47,7 @@ mod engine;
 mod local_scan;
 mod model;
 mod planner;
-use engine::SYNC_CONCURRENCY;
+pub(crate) use engine::SYNC_CONCURRENCY;
 pub(crate) use engine::{SyncMsg, spawn};
 #[cfg(test)]
 use engine::{classify, event_target, settle_with};

crates/pdfs-fuse/src/sync/engine.rs

@@ -22,7 +22,7 @@ const POLL_INTERVAL: Duration = Duration::from_secs(120);
 /// per-item work is a network round-trip, so a folder with thousands of files is
 /// unusable done one-at-a-time; this bounds concurrency like the bulk-upload
 /// engine does.
-pub(super) const SYNC_CONCURRENCY: usize = 8;
+pub(crate) const SYNC_CONCURRENCY: usize = 8;
 
 /// A message to the sync engine's control thread.
 pub(crate) enum SyncMsg {

crates/pdfs-fuse/src/transfers.rs

@@ -125,6 +125,11 @@ impl TransferRegistry {
         self.download_limit.set(download);
     }
 
+    /// The upload cap in bytes per second, `0` when there is none.
+    pub fn upload_cap(&self) -> u64 {
+        self.upload_limit.rate.load(Ordering::Relaxed)
+    }
+
     fn limit(&self, direction: TransferDirection) -> &RateLimit {
         match direction {
             TransferDirection::Upload => &self.upload_limit,