fix(core,fuse): back failed uploads off properly and spread their retries

A reconnect made every failed op due at once, quota and access refusals
included, and ops that failed together retried in the same second. The
doubling also stopped at 128 s, short of the 300 s maximum. Reconnects
now move up only unrefused ops, and the backoff reaches its maximum with
a per-op spread of up to a fifth (B237).

crates/pdfs-core/src/db/ops.rs

@@ -1483,12 +1483,26 @@ impl Db {
     }
 
     /// [`retry_op_now`](Self::retry_op_now) for every op that has failed at
-    /// least once. Returns how many rows changed.
+    /// least once, for the user's "Retry all". Returns how many rows changed.
     pub fn retry_failed_ops_now(&self, now: i64) -> Result<usize> {
+        self.retry_ops_now(now, "")
+    }
+
+    /// [`retry_failed_ops_now`](Self::retry_failed_ops_now) for the ops Drive
+    /// did not refuse, for a reconnect. A link coming back changes nothing
+    /// about a full quota or a missing permission, so those keep their backoff
+    /// instead of all being sent again at once.
+    pub fn retry_unrefused_ops_now(&self, now: i64) -> Result<usize> {
+        self.retry_ops_now(now, "AND issue IS NULL")
+    }
+
+    fn retry_ops_now(&self, now: i64, filter: &str) -> Result<usize> {
         let conn = self.conn.lock();
         let changed = conn.execute(
-            "UPDATE pending_op SET next_attempt_at = ?1
-             WHERE attempts > 0 AND next_attempt_at > ?1 AND next_attempt_at < ?2",
+            &format!(
+                "UPDATE pending_op SET next_attempt_at = ?1
+                 WHERE attempts > 0 AND next_attempt_at > ?1 AND next_attempt_at < ?2 {filter}"
+            ),
             params![now, PARK_UNTIL],
         )?;
         Ok(changed)

crates/pdfs-core/src/db/tests.rs

@@ -7531,6 +7531,32 @@ fn a_refused_change_names_its_node_by_local_id() {
     );
 }
 
+/// A reconnect moves up the ops that failed for want of a link. One Drive
+/// refused, for quota or access, keeps its backoff: the link coming back
+/// changes nothing about it. "Retry all" still moves both.
+#[test]
+fn a_reconnect_leaves_refused_ops_to_their_backoff() {
+    let db = Db::open_in_memory().unwrap();
+    let offline = queued_op(&db, "vol~a", 2, 90_000);
+    let refused = queued_op(&db, "vol~b", 2, 90_000);
+    db.set_op_issue(refused, Some("quota")).unwrap();
+
+    assert_eq!(db.retry_unrefused_ops_now(5_000).unwrap(), 1);
+    let due = |id: i64| {
+        db.pending_ops()
+            .unwrap()
+            .into_iter()
+            .find(|op| op.id == id)
+            .unwrap()
+            .next_attempt_at
+    };
+    assert_eq!(due(offline), 5_000);
+    assert_eq!(due(refused), 90_000);
+
+    assert_eq!(db.retry_failed_ops_now(5_000).unwrap(), 1);
+    assert_eq!(due(refused), 5_000);
+}
+
 #[test]
 fn retry_all_touches_only_ops_that_have_failed() {
     let db = Db::open_in_memory().unwrap();

crates/pdfs-fuse/src/drain.rs

@@ -389,13 +389,7 @@ impl Core {
                 // A name a change of ours is about to free is a wait, not a
                 // failure: the trash or rename holding it lands in moments.
                 let held = waits_for_name(e.as_ref());
-                let backoff = if held {
-                    DRAIN_BACKOFF_MIN
-                } else {
-                    DRAIN_BACKOFF_MIN
-                        .saturating_mul(1u32 << attempts.min(6))
-                        .min(DRAIN_BACKOFF_MAX)
-                };
+                let backoff = drain_backoff(attempts, held, op.id);
                 warn!(uid = %op.uid, attempts, error = %err_str, "pending upload failed; will retry");
                 if let Err(e) = self.db.record_attempt_failure(
                     op.id,
@@ -2787,6 +2781,25 @@ fn waits_for_name(e: &(dyn std::error::Error + 'static)) -> bool {
     std::iter::successors(Some(e), |e| e.source()).any(|e| e.is::<NameHeld>())
 }
 
+/// How long an op waits after its `attempts`-th failure: doubling from
+/// [`DRAIN_BACKOFF_MIN`] up to [`DRAIN_BACKOFF_MAX`], plus up to a fifth more,
+/// different for each op (`seed`), so that ops that failed together are not
+/// all retried in the same second. One that waits for a name (`held`) waits
+/// the minimum.
+///
+/// The doubling stopped at 2^6, 128 s, so the maximum was never reached.
+fn drain_backoff(attempts: i64, held: bool, seed: i64) -> Duration {
+    if held {
+        return DRAIN_BACKOFF_MIN;
+    }
+    let base = DRAIN_BACKOFF_MIN
+        .saturating_mul(1u32 << attempts.clamp(0, 8))
+        .min(DRAIN_BACKOFF_MAX);
+    // A cheap, stable spread: the same op and attempt always wait the same.
+    let spread = (seed.wrapping_mul(0x9E37_79B9).wrapping_add(attempts) as u64 % 21) as u32;
+    base + base * spread / 100
+}
+
 /// How many attempts an op waits for a rename or a create on the wire to free
 /// the name it wants. Two files swapping names each wait for the other, so the
 /// wait has to end; a conflict name ends it.
@@ -3386,6 +3399,22 @@ mod tests {
         }
     }
 
+    #[test]
+    fn a_failing_op_backs_off_to_the_maximum_with_a_spread() {
+        let at = |attempts| drain_backoff(attempts, false, 7);
+        assert!(at(1) >= DRAIN_BACKOFF_MIN * 2);
+        assert!(at(2) > at(1));
+        // Reached, which the doubling capped at 2^6 never did.
+        assert!(at(8) >= DRAIN_BACKOFF_MAX);
+        assert!(at(30) <= DRAIN_BACKOFF_MAX + DRAIN_BACKOFF_MAX / 5);
+        // Ops that failed together do not all come back together.
+        let waits: std::collections::HashSet<_> =
+            (1..=20).map(|id| drain_backoff(8, false, id)).collect();
+        assert!(waits.len() > 5, "{waits:?}");
+        // A wait for a name stays short however often it repeats.
+        assert_eq!(drain_backoff(30, true, 7), DRAIN_BACKOFF_MIN);
+    }
+
     #[test]
     fn only_a_fresh_empty_twin_of_the_same_kind_is_adopted() {
         // The op was queued at t=1000 s, after a create that got no answer.

crates/pdfs-fuse/src/link.rs

@@ -280,8 +280,9 @@ impl Core {
         self.relist_offline_listings();
         // Anything written while offline is queued and waiting on exactly this,
         // including ops that failed on the way down and are sitting out a
-        // backoff sized for a server that keeps refusing.
-        if let Err(e) = self.db.retry_failed_ops_now(crate::now_millis()) {
+        // backoff sized for a server that keeps refusing. Ops Drive refused
+        // keep theirs: the link was not why.
+        if let Err(e) = self.db.retry_unrefused_ops_now(crate::now_millis()) {
             warn!(error = %e, "moving backed-off ops up after a reconnect failed");
         }
         self.wake_drain();