fix(drive) locks + 1.7.1

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

@@ -9,7 +9,7 @@ use super::Db;
 use crate::Result;
 
 /// Current schema version. Bump on every forward migration added below.
-pub(super) const SCHEMA_VERSION: i64 = 25;
+pub(super) const SCHEMA_VERSION: i64 = 26;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -161,6 +161,27 @@ impl Db {
                 tx.execute_batch(MIGRATION_V25)?;
             }
         }
+        if current < 26 {
+            // Same guard as V22: a fixture that never exercised the queue may
+            // not have `pending_op`, and altering an absent table is an error
+            // rather than a no-op.
+            // A fixture built from the current schema and then rewound to an
+            // older `schema_version` already has the column, and `ADD COLUMN`
+            // has no `IF NOT EXISTS`.
+            let has_column: bool = tx.query_row(
+                "SELECT COUNT(*) FROM pragma_table_info('pending_op') WHERE name = 'claimed_at'",
+                [],
+                |row| row.get::<_, i64>(0),
+            )? > 0;
+            let has_ops: bool = tx.query_row(
+                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'pending_op'",
+                [],
+                |row| row.get::<_, i64>(0),
+            )? > 0;
+            if has_ops && !has_column {
+                tx.execute_batch(MIGRATION_V26)?;
+            }
+        }
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -725,3 +746,26 @@ UPDATE nodes
 
 DROP TABLE node_paths;
 ";
+
+/// V26: `pending_op.claimed_at` — the drain's claim column.
+///
+/// The drain was one thread taking one op at a time, so a 10 GiB upload held
+/// every queued rename and small write behind it for as long as it ran. Several
+/// workers can share the queue only if a claim is visible to the others: this
+/// column is what makes "already being drained by someone" a fact about the row
+/// rather than about one thread's local variable.
+///
+/// `0` is unclaimed; anything else is the ms at which a worker took it. A
+/// claim is process-local state that outlives nothing — the single-writer lock
+/// means a non-zero value found at open belongs to a crashed run, and
+/// [`Db::clear_op_claims`](super::Db::clear_op_claims) drops the lot.
+///
+/// The index is partial so it costs nothing on the ordinary row: only the
+/// handful of claimed rows are in it, which is exactly the set the claim query
+/// has to subtract to keep two workers off one node.
+const MIGRATION_V26: &str = "
+ALTER TABLE pending_op ADD COLUMN claimed_at INTEGER NOT NULL DEFAULT 0;
+
+CREATE INDEX IF NOT EXISTS idx_pending_op_claimed
+    ON pending_op(uid) WHERE claimed_at <> 0;
+";

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

@@ -230,7 +230,7 @@ impl Db {
             }
             Err(e) => return Err(e),
         };
-        Ok(Self {
+        let db = Self {
             conn: Mutex::new(conn),
             // No pool for `:memory:`. A second connection to that name is a
             // second, empty database rather than another view of this one, so
@@ -240,7 +240,17 @@ impl Db {
                 idle: Mutex::new(Vec::new()),
             }),
             _single_writer: single_writer,
-        })
+        };
+        // A drain claim names a worker thread in the process that took it, and
+        // the single-writer lock above says that process is not running. Any
+        // claim still on disk is therefore a crashed run's, and leaving it would
+        // hide those ops from every worker for good.
+        match db.clear_op_claims() {
+            Ok(0) => {}
+            Ok(n) => tracing::info!(ops = n, "released drain claims left by a previous run"),
+            Err(e) => tracing::warn!(error = %e, "clearing stale drain claims failed"),
+        }
+        Ok(db)
     }
 
     /// A connection for a `SELECT`-only method.

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

@@ -512,17 +512,115 @@ impl Db {
         Ok(op)
     }
 
+    /// Take the oldest due op for this worker, marking it claimed so no other
+    /// drain worker takes it, or `None` when there is nothing to do.
+    ///
+    /// The drain is several workers over one queue, which is only safe because
+    /// the claim is a column: a row is offered to exactly one worker, and the
+    /// worker that took it is the only one that may retire, defer or fail it.
+    ///
+    /// Two exclusions on top of [`next_due_op`](Self::next_due_op)'s:
+    ///
+    /// * `claimed_at = 0` — not already in flight somewhere else.
+    /// * no *other* claimed op shares this `uid`. Ordering only has to hold per
+    ///   node, but there it has to hold absolutely: a queued rename and a queued
+    ///   revision of one file must land in the order they were made, and two
+    ///   workers on one node would also race over its staged blob and its
+    ///   pending entry. `ORDER BY id` plus this exclusion gives per-uid serial
+    ///   order for free — the older op is claimed first and the newer one is
+    ///   invisible until it retires.
+    ///
+    /// Select and mark are one transaction, so two workers cannot both see the
+    /// row as unclaimed.
+    pub fn claim_next_due_op(&self, now: i64) -> Result<Option<PendingOp>> {
+        let mut conn = self.conn.lock();
+        let tx = conn.transaction()?;
+        let op = {
+            let mut stmt = tx.prepare_cached(&format!(
+                "SELECT id, kind, uid, parent_uid, name, blob_path, meta_json, created_at,
+                        attempts, last_error, next_attempt_at
+                 FROM pending_op
+                 WHERE next_attempt_at <= ?1
+                   AND claimed_at = 0
+                   AND (parent_uid IS NULL OR substr(parent_uid, 1, {n}) <> '{v}~')
+                   AND uid NOT IN (SELECT uid FROM pending_op WHERE claimed_at <> 0)
+                 ORDER BY id LIMIT 1",
+                v = LOCAL_VOLUME,
+                n = LOCAL_VOLUME.len() + 1,
+            ))?;
+            stmt.query_row(params![now], |r| {
+                Ok(PendingOp {
+                    id: r.get(0)?,
+                    kind: r.get(1)?,
+                    uid: r.get(2)?,
+                    parent_uid: r.get(3)?,
+                    name: r.get(4)?,
+                    blob_path: r.get(5)?,
+                    meta_json: r.get(6)?,
+                    created_at: r.get(7)?,
+                    attempts: r.get(8)?,
+                    last_error: r.get(9)?,
+                    next_attempt_at: r.get(10)?,
+                })
+            })
+            .optional()?
+        };
+        if let Some(op) = op.as_ref() {
+            tx.execute(
+                "UPDATE pending_op SET claimed_at = ?2 WHERE id = ?1",
+                params![op.id, now.max(1)],
+            )?;
+        }
+        tx.commit()?;
+        Ok(op)
+    }
+
+    /// Give a claimed op back to the queue, whatever became of the attempt.
+    ///
+    /// Called on every path out of a drain attempt rather than only the failing
+    /// ones: a handler that retires its own row leaves nothing to release (the
+    /// `UPDATE` matches nothing), and a handler that returns without retiring
+    /// *and* without failing would otherwise leave the row claimed by a worker
+    /// that has moved on — invisible to every worker, forever.
+    pub fn release_op_claim(&self, id: i64) -> Result<()> {
+        let conn = self.conn.lock();
+        conn.execute(
+            "UPDATE pending_op SET claimed_at = 0 WHERE id = ?1",
+            params![id],
+        )?;
+        Ok(())
+    }
+
+    /// Drop every claim in the table.
+    ///
+    /// A claim describes a worker in *this* process, so one surviving on disk
+    /// means the previous run died holding it. The single-writer lock
+    /// ([`Db::open`]) is what makes that inference safe: nobody else can be
+    /// draining this queue while we are opening it.
+    pub fn clear_op_claims(&self) -> Result<usize> {
+        let conn = self.conn.lock();
+        Ok(conn.execute(
+            "UPDATE pending_op SET claimed_at = 0 WHERE claimed_at <> 0",
+            [],
+        )?)
+    }
+
     /// The earliest `next_attempt_at` among all queued ops, or `None` if the
     /// queue is empty. Used by the drain loop to sleep exactly until the next
     /// debounced or backed-off op becomes eligible rather than waiting the full
     /// idle-poll interval.
+    ///
+    /// Claimed ops are excluded: another worker is already on them, so they are
+    /// not work this one is waiting for, and counting them would have an idle
+    /// worker spin on a row it cannot have.
     pub fn earliest_due_at(&self) -> Result<Option<i64>> {
         let conn = self.read();
         let ts: Option<i64> = conn
             .query_row(
                 &format!(
                     "SELECT MIN(next_attempt_at) FROM pending_op \
-                     WHERE parent_uid IS NULL OR substr(parent_uid, 1, {n}) <> '{v}~'",
+                     WHERE claimed_at = 0 \
+                       AND (parent_uid IS NULL OR substr(parent_uid, 1, {n}) <> '{v}~')",
                     v = LOCAL_VOLUME,
                     n = LOCAL_VOLUME.len() + 1,
                 ),

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

@@ -3545,3 +3545,114 @@ fn a_read_does_not_wait_for_the_write_connection() {
     let _ = std::fs::remove_file(&path);
     let _ = std::fs::remove_file(format!("{}.lock", path.display()));
 }
+
+/// §5: the drain is several workers over one queue, and the claim column is the
+/// only thing keeping them off each other's work. Two properties matter — no
+/// row is ever handed to two workers, and no *node* is ever worked by two
+/// workers even though its ops are separate rows.
+#[test]
+fn a_claimed_op_is_invisible_to_every_other_worker() {
+    let db = Db::open_in_memory().unwrap();
+    let root = uid("root").to_string();
+    for i in 0..3 {
+        db.enqueue_op(&bulk_op(i, &root)).unwrap();
+    }
+
+    let first = db.claim_next_due_op(10).unwrap().expect("an op is due");
+    let second = db
+        .claim_next_due_op(10)
+        .unwrap()
+        .expect("another op is due");
+    let third = db
+        .claim_next_due_op(10)
+        .unwrap()
+        .expect("a third op is due");
+    assert!(
+        db.claim_next_due_op(10).unwrap().is_none(),
+        "a fourth claim must find nothing left: every queued op is taken"
+    );
+
+    let mut ids = [first.id, second.id, third.id];
+    ids.sort_unstable();
+    assert_eq!(
+        ids,
+        [1, 2, 3],
+        "each worker must get a distinct row, oldest first"
+    );
+
+    // Releasing puts one back, and it is the one that comes out next.
+    db.release_op_claim(second.id).unwrap();
+    let again = db.claim_next_due_op(10).unwrap().expect("the released op");
+    assert_eq!(again.id, second.id);
+}
+
+#[test]
+fn two_workers_never_share_a_node() {
+    let db = Db::open_in_memory().unwrap();
+    let root = uid("root").to_string();
+    // Two ops of different kinds against one node — a queued rename and a
+    // queued revision, which is an ordinary `mv` plus a save.
+    let mut revision = bulk_op(0, &root);
+    revision.parent_uid = None;
+    db.enqueue_op(&revision).unwrap();
+    let rename = PendingOp {
+        kind: OP_RENAME.to_string(),
+        parent_uid: Some(root.clone()),
+        blob_path: None,
+        meta_json: None,
+        ..bulk_op(0, &root)
+    };
+    db.enqueue_op(&rename).unwrap();
+
+    let claimed = db.claim_next_due_op(10).unwrap().expect("an op is due");
+    assert_eq!(claimed.kind, OP_REVISION, "the older op goes first");
+    assert!(
+        db.claim_next_due_op(10).unwrap().is_none(),
+        "the second op targets the same node, so it must wait for the first: \
+         two workers on one uid would race over its staged blob and land its \
+         changes out of order"
+    );
+
+    db.release_op_claim(claimed.id).unwrap();
+    db.delete_op(claimed.id).unwrap();
+    assert_eq!(
+        db.claim_next_due_op(10).unwrap().map(|o| o.kind),
+        Some(OP_RENAME.to_string()),
+        "once the revision retires, the rename is claimable"
+    );
+}
+
+#[test]
+fn claims_do_not_survive_the_process_that_took_them() {
+    let db = Db::open_in_memory().unwrap();
+    let root = uid("root").to_string();
+    for i in 0..2 {
+        db.enqueue_op(&bulk_op(i, &root)).unwrap();
+    }
+    db.claim_next_due_op(10).unwrap().unwrap();
+    db.claim_next_due_op(10).unwrap().unwrap();
+    assert!(db.claim_next_due_op(10).unwrap().is_none());
+
+    // What `Db::open` does after the single-writer lock says the previous run
+    // is gone. Without it those ops are invisible to every worker forever.
+    assert_eq!(db.clear_op_claims().unwrap(), 2);
+    assert!(
+        db.claim_next_due_op(10).unwrap().is_some(),
+        "a crashed run's claims must not park its queue"
+    );
+}
+
+#[test]
+fn an_idle_worker_does_not_wait_on_an_op_another_worker_has() {
+    let db = Db::open_in_memory().unwrap();
+    let root = uid("root").to_string();
+    db.enqueue_op(&bulk_op(0, &root)).unwrap();
+
+    assert_eq!(db.earliest_due_at().unwrap(), Some(0));
+    db.claim_next_due_op(10).unwrap().unwrap();
+    assert_eq!(
+        db.earliest_due_at().unwrap(),
+        None,
+        "an op somebody else is draining is not work this worker is waiting for"
+    );
+}

crates/pdfs-fuse/src/drain.rs

@@ -20,10 +20,14 @@
 //! are kept as a conflict copy rather than discarded — see
 //! [`Core::revision_conflict`].
 
+use std::collections::HashMap;
 use std::fs::File;
 use std::path::{Path, PathBuf};
-use std::sync::atomic::Ordering;
-use std::time::Duration;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::{Duration, Instant};
+
+use parking_lot::Mutex;
 
 use pdfs_core::batch;
 use pdfs_core::cache::{Baseline, StagedWrite};
@@ -36,13 +40,59 @@ use tracing::{debug, error, info, warn};
 use super::state::Intervals;
 use super::transfers::CountingReader;
 use super::{
-    Core, DRAIN_BACKOFF_MAX, DRAIN_BACKOFF_MIN, DRAIN_IDLE_POLL, ROOT_INO, conflict_name,
-    is_already_exists, is_gone, is_local_uid_str, media_type_for, node_revision_id, node_size,
-    now_millis, now_secs, parse_node_uid,
+    Core, DRAIN_BACKOFF_MAX, DRAIN_BACKOFF_MIN, DRAIN_IDLE_POLL, DRAIN_REVISION_DEBOUNCE,
+    DRAIN_REVISION_DEBOUNCE_MAX, ROOT_INO, UPLOAD_TIME_MEMORY, conflict_name, is_already_exists,
+    is_gone, is_local_uid_str, media_type_for, node_revision_id, node_size, now_millis, now_secs,
+    parse_node_uid,
 };
 
 const DRAIN_ACCESS_RECHECK: Duration = Duration::from_secs(5);
 
+/// A registered cancellation flag for one in-flight upload, deregistered when
+/// dropped. See [`Core::begin_cancellable_upload`].
+struct UploadCancel {
+    registry: Arc<Mutex<HashMap<NodeUid, Arc<AtomicBool>>>>,
+    uid: NodeUid,
+    flag: Arc<AtomicBool>,
+}
+
+impl UploadCancel {
+    /// Whether this upload was told to stop. Distinguishes an upload we
+    /// abandoned on purpose from one the network broke.
+    fn cancelled(&self) -> bool {
+        self.flag.load(Ordering::Relaxed)
+    }
+
+    fn flag(&self) -> Arc<AtomicBool> {
+        self.flag.clone()
+    }
+}
+
+impl Drop for UploadCancel {
+    fn drop(&mut self) {
+        let mut registry = self.registry.lock();
+        // By identity: a supersede racing this upload's last bytes may already
+        // have registered the next one under the same uid.
+        if registry
+            .get(&self.uid)
+            .is_some_and(|current| Arc::ptr_eq(current, &self.flag))
+        {
+            registry.remove(&self.uid);
+        }
+    }
+}
+
+/// The debounce arithmetic of [`Core::revision_debounce`], without the map:
+/// the last measured upload time for this node, bounded on both sides.
+///
+/// A node nobody has uploaded yet gets the fixed grace period, which is the
+/// same answer the fixed debounce always gave.
+fn adaptive_debounce(measured: Option<Duration>) -> Duration {
+    measured
+        .unwrap_or(DRAIN_REVISION_DEBOUNCE)
+        .clamp(DRAIN_REVISION_DEBOUNCE, DRAIN_REVISION_DEBOUNCE_MAX)
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum DrainDisposition {
     Applied,
@@ -127,24 +177,57 @@ impl Core {
     /// that is due — one file wedged against a folder that no longer exists must
     /// not hold up an unrelated upload behind it. The worker only blocks once
     /// nothing is due at all.
-    pub(crate) fn run_pending_drain(&self) {
+    ///
+    /// Several of these run at once ([`DRAIN_WORKERS`]), sharing the queue
+    /// through [`Db::claim_next_due_op`]: the alternative was that a 10 GiB
+    /// upload held every queued rename, trash and small write behind it for as
+    /// long as it took. Ordering only has to hold *per node*, and the claim
+    /// query is what guarantees it — no two workers are ever on one uid.
+    ///
+    /// `primary` marks the one worker that also runs the queue's idle chores.
+    /// They are cheap, but they are not per-op work and doing them in every
+    /// worker would just multiply them.
+    ///
+    /// [`DRAIN_WORKERS`]: super::DRAIN_WORKERS
+    /// [`Db::claim_next_due_op`]: pdfs_core::db::Db::claim_next_due_op
+    pub(crate) fn run_pending_drain(&self, primary: bool) {
         loop {
             let now = now_millis();
-            // One row, chosen by the database. Reading the whole queue to pick
-            // one op made a long queue quadratic to drain, and held the shared
-            // connection — and so every FUSE metadata call — for the duration.
-            let due = self.db.next_due_op(now).unwrap_or_default();
-
-            let Some(op) = due.filter(|_| self.online.load(Ordering::Relaxed)) else {
-                // Nothing to upload is exactly when the connection is free, so
-                // this is where the read path's buffered LRU touches get written
-                // (`ContentCache::flush_touches`). Cheap and a no-op when empty.
-                self.cache.flush_touches();
-                self.recover_fsynced_writes();
-                // A debounced or backed-off op may be waiting: sleep only
-                // until it becomes due rather than the full idle-poll.
-                self.wait_for_drain_work_or_due();
-                continue;
+            // One row, chosen and *claimed* by the database. Reading the whole
+            // queue to pick one op made a long queue quadratic to drain, and
+            // held the shared connection — and so every FUSE metadata call —
+            // for the duration.
+            let due = match self.db.claim_next_due_op(now) {
+                Ok(due) => due,
+                Err(error) => {
+                    error!(%error, "claiming the next pending operation failed");
+                    self.wait_for_drain_work();
+                    continue;
+                }
+            };
+            let online = self.online.load(Ordering::Relaxed);
+            let op = match due {
+                Some(op) if online => op,
+                idle => {
+                    // Claimed but offline: hand it straight back, or it stays
+                    // invisible to every worker until this one happens to be
+                    // the next to notice the reconnect.
+                    if let Some(op) = idle {
+                        self.release_claim(&op);
+                    }
+                    if primary {
+                        // Nothing to upload is exactly when the connection is
+                        // free, so this is where the read path's buffered LRU
+                        // touches get written (`ContentCache::flush_touches`).
+                        // Cheap and a no-op when empty.
+                        self.cache.flush_touches();
+                        self.recover_fsynced_writes();
+                    }
+                    // A debounced or backed-off op may be waiting: sleep only
+                    // until it becomes due rather than the full idle-poll.
+                    self.wait_for_drain_work_or_due();
+                    continue;
+                }
             };
             let outcome = run_authorized_drain(
                 &op,
@@ -159,6 +242,7 @@ impl Core {
                         %error,
                         "deferring access-blocked pending operation failed"
                     );
+                    self.release_claim(&op);
                     self.wait_for_drain_work();
                 } else {
                     debug!(
@@ -166,6 +250,7 @@ impl Core {
                         next_attempt_at,
                         "pending operation deferred until access is writable"
                     );
+                    self.release_claim(&op);
                 }
                 continue;
             }
@@ -188,9 +273,103 @@ impl Core {
                     now_millis() + backoff.as_millis() as i64,
                 ) {
                     error!(uid = %op.uid, error = %e, "recording a drain failure failed");
+                    self.release_claim(&op);
                     self.wait_for_drain_work();
+                    continue;
                 }
             }
+            // Unconditionally, on every path: a handler that retired its own row
+            // leaves nothing to release, and one that returned without retiring
+            // must not leave the row claimed by a worker that has moved on.
+            self.release_claim(&op);
+        }
+    }
+
+    /// Hand a claimed op back to the queue, logging rather than propagating —
+    /// there is no caller left to propagate to, and the drain must not stop.
+    fn release_claim(&self, op: &PendingOp) {
+        if let Err(error) = self.db.release_op_claim(op.id) {
+            error!(uid = %op.uid, %error, "releasing a drain claim failed");
+        }
+    }
+
+    /// How long a freshly queued revision of `uid` should wait before a worker
+    /// may pick it up.
+    ///
+    /// [`DRAIN_REVISION_DEBOUNCE`] exists for a tool that preallocates a file
+    /// and then writes it, where two seconds is the right scale. It is the wrong
+    /// scale for a large file on a slow link: an editor saving a 200 MB project
+    /// every ten seconds queues a write, the drain starts sending it, the next
+    /// save supersedes it mid-flight, and the file never finishes uploading
+    /// while the user keeps working. Waiting roughly as long as the last upload
+    /// of *this* node took moves that supersede into the queue, where replacing
+    /// a row costs nothing.
+    ///
+    /// Never shorter than the fixed debounce and never longer than
+    /// [`DRAIN_REVISION_DEBOUNCE_MAX`]: a node whose upload is genuinely slow
+    /// still has to reach Drive, and a single pathological stall must not park
+    /// it.
+    ///
+    /// [`DRAIN_REVISION_DEBOUNCE`]: super::DRAIN_REVISION_DEBOUNCE
+    /// [`DRAIN_REVISION_DEBOUNCE_MAX`]: super::DRAIN_REVISION_DEBOUNCE_MAX
+    pub(crate) fn revision_debounce(&self, uid: &NodeUid) -> Duration {
+        let measured = self
+            .upload_times
+            .lock()
+            .get(&uid.to_string())
+            .map(|&(ms, _)| Duration::from_millis(ms));
+        adaptive_debounce(measured)
+    }
+
+    /// Remember how long an upload of `uid` took, for the next write's
+    /// debounce.
+    ///
+    /// Only a *completed* upload is a measurement. A cancelled or failed one
+    /// says how long it ran before it stopped, which is not how long the file
+    /// takes to send, and feeding that back would let one network fault widen
+    /// the debounce for good.
+    fn record_upload_time(&self, uid: &NodeUid, took: Duration) {
+        let mut times = self.upload_times.lock();
+        let now = now_millis();
+        if times.len() >= UPLOAD_TIME_MEMORY
+            && !times.contains_key(&uid.to_string())
+            && let Some(oldest) = times
+                .iter()
+                .min_by_key(|(_, (_, at))| *at)
+                .map(|(k, _)| k.clone())
+        {
+            times.remove(&oldest);
+        }
+        times.insert(uid.to_string(), (took.as_millis() as u64, now));
+    }
+
+    /// Register a cancellation flag for an upload of `uid` that is about to
+    /// start, returning a guard that deregisters it.
+    ///
+    /// Deregistration is by identity rather than by key: a supersede that
+    /// arrives just as this upload finishes may already have registered the
+    /// *next* upload's flag under the same uid, and removing that one would
+    /// leave the newer upload uncancellable.
+    fn begin_cancellable_upload(&self, uid: &NodeUid) -> UploadCancel {
+        let flag = Arc::new(AtomicBool::new(false));
+        self.upload_cancel.lock().insert(uid.clone(), flag.clone());
+        UploadCancel {
+            registry: self.upload_cancel.clone(),
+            uid: uid.clone(),
+            flag,
+        }
+    }
+
+    /// Tell an upload of `uid` that is on the wire to stop, if there is one.
+    ///
+    /// Called when a newer revision of the file is queued. The upload does not
+    /// stop instantly — it stops when the SDK next asks its reader for bytes,
+    /// which for a 4 MiB block is soon enough to matter and late enough to be
+    /// safe: the staged blob it is reading stays open until it lets go.
+    pub(crate) fn cancel_upload(&self, uid: &NodeUid) {
+        if let Some(flag) = self.upload_cancel.lock().get(uid) {
+            flag.store(true, Ordering::Relaxed);
+            debug!(%uid, "cancelling a superseded upload");
         }
     }
 
@@ -907,15 +1086,37 @@ impl Core {
         let guard =
             self.transfers
                 .begin(&name, meta.uid.clone(), TransferDirection::Upload, meta.len);
-        let reader = CountingReader::new(File::open(&blob)?, &guard);
-        self.rt.block_on(self.client.upload_new_revision_from(
+        // Registered before the first byte moves, so a write that lands while
+        // this is on the wire can stop it rather than paying for a revision the
+        // op it queued will immediately replace.
+        let cancel = self.begin_cancellable_upload(&uid);
+        let reader = CountingReader::new(File::open(&blob)?, &guard).with_cancel(cancel.flag());
+        let started = Instant::now();
+        let sent = self.rt.block_on(self.client.upload_new_revision_from(
             &uid,
             reader,
             meta.len as i64,
             Vec::new(),
             None,
-        ))?;
+        ));
+        let took = started.elapsed();
+        if let Err(e) = sent {
+            if cancel.cancelled() {
+                // Not a failure, and not this op's row to retire: the write that
+                // cancelled us deleted it and took ownership of the staged blob
+                // when it superseded this revision. Leaving the blob alone here
+                // is the whole point — it is either already unlinked or it
+                // belongs to the newer op.
+                info!(%uid, "upload abandoned; a newer write superseded it");
+                return Ok(());
+            }
+            return Err(e.into());
+        }
+        drop(cancel);
         drop(guard);
+        // Only a completed upload times how long this file takes to send, which
+        // is what the next write's debounce is scaled from.
+        self.record_upload_time(&uid, took);
 
         // Retire the op before dropping the blob: a crash between the two leaves
         // an orphaned file (harmless), whereas the reverse would leave a queued
@@ -1413,3 +1614,42 @@ mod tests {
         assert!(!is_own_self_supersede(true, None, None));
     }
 }
+
+#[cfg(test)]
+mod debounce_tests {
+    use super::*;
+
+    #[test]
+    fn a_node_never_uploaded_keeps_the_fixed_grace_period() {
+        assert_eq!(adaptive_debounce(None), DRAIN_REVISION_DEBOUNCE);
+    }
+
+    #[test]
+    fn a_fast_upload_does_not_shorten_the_grace_period() {
+        // The 2 seconds are not about upload time at all — they are the window
+        // in which a preallocate-then-write tool supersedes its own first
+        // close. A file that sends in 50 ms must still get them.
+        assert_eq!(
+            adaptive_debounce(Some(Duration::from_millis(50))),
+            DRAIN_REVISION_DEBOUNCE
+        );
+    }
+
+    #[test]
+    fn a_slow_upload_widens_the_grace_period_to_match() {
+        // The case the fixed debounce got wrong: saves arriving faster than the
+        // upload completes, each superseding one already on the wire.
+        assert_eq!(
+            adaptive_debounce(Some(Duration::from_secs(25))),
+            Duration::from_secs(25)
+        );
+    }
+
+    #[test]
+    fn a_pathological_upload_cannot_park_a_node() {
+        assert_eq!(
+            adaptive_debounce(Some(Duration::from_secs(6000))),
+            DRAIN_REVISION_DEBOUNCE_MAX
+        );
+    }
+}

crates/pdfs-fuse/src/lib.rs

@@ -166,6 +166,30 @@ const DRAIN_IDLE_POLL: Duration = Duration::from_secs(30);
 /// creates a conflict copy of itself. Holding the op for a short window gives
 /// the follow-up write time to supersede it.
 const DRAIN_REVISION_DEBOUNCE: Duration = Duration::from_secs(2);
+/// Ceiling on the *adaptive* part of that grace period.
+///
+/// The fixed 2 seconds is enough for a preallocate-then-write tool, and wrong
+/// for a file that takes half a minute to upload: an editor saving every ten
+/// seconds queues a write, watches it start uploading, and supersedes it — each
+/// save uploading bytes that the next one replaces, and never catching up.
+/// Widening the debounce toward how long this node's last upload actually took
+/// makes the supersede happen in the queue, where it is free, instead of on the
+/// wire. Bounded so a file on a slow link still reaches Drive in a minute rather
+/// than deferring indefinitely, and so a one-off stall cannot park a node.
+const DRAIN_REVISION_DEBOUNCE_MAX: Duration = Duration::from_secs(60);
+/// How many nodes' measured upload times to remember for that. Bounds a map
+/// that would otherwise grow with every file ever written; the debounce falls
+/// back to [`DRAIN_REVISION_DEBOUNCE`] for anything evicted, which is the
+/// behaviour a node that has never been uploaded gets anyway.
+const UPLOAD_TIME_MEMORY: usize = 512;
+/// Threads draining the pending-op queue.
+///
+/// One meant a single 10 GiB upload blocked every queued rename, trash and
+/// small write behind it. Ordering only has to hold per node — enforced by the
+/// claim query, not by the thread count — so the rest is throughput. Kept small:
+/// these are uploads, and past a handful they compete for the same uplink and
+/// for the SDK's own per-request concurrency.
+const DRAIN_WORKERS: usize = 3;
 
 const ONLINE_PROBE_MIN: Duration = Duration::from_secs(5);
 const ONLINE_PROBE_MAX: Duration = Duration::from_secs(300);
@@ -423,9 +447,23 @@ struct Core {
     /// Nodes removed through this daemon which a briefly stale remote listing
     /// may still return. Uids are not reused; an explicit restore clears one.
     hidden: Arc<Mutex<HashSet<NodeUid>>>,
-    /// Nudges the drain worker: set true and notify to have it re-examine the
-    /// queue instead of waiting out its backoff.
+    /// Nudges the drain workers: set true and notify to have them re-examine
+    /// the queue instead of waiting out their backoff.
     drain_wake: Arc<(Mutex<bool>, Condvar)>,
+    /// How long this node's last upload actually took, in milliseconds, with
+    /// the time it was measured for eviction. Feeds
+    /// [`Core::revision_debounce`]: a file that takes 30 seconds to send needs
+    /// a grace period on that scale, or every save supersedes an upload that
+    /// was already on the wire. Bounded by [`UPLOAD_TIME_MEMORY`].
+    upload_times: Arc<Mutex<HashMap<String, (u64, i64)>>>,
+    /// Cancellation flags for uploads currently on the wire, keyed by node.
+    ///
+    /// Set when a newer revision of the same file is queued: the bytes in
+    /// flight have just been superseded, and the only thing finishing them
+    /// achieves is spending the user's uplink on a revision the next op
+    /// replaces. Read by the upload's own reader ([`CountingReader`]), which is
+    /// the one thing the SDK calls often enough to notice.
+    upload_cancel: Arc<Mutex<HashMap<NodeUid, Arc<AtomicBool>>>>,
     /// True while a background refresh of the photos timeline (resp. the trash) is
     /// already running, so a burst of page requests against a stale listing kicks
     /// off one refresh rather than one per request.
@@ -2756,6 +2794,12 @@ impl Core {
             }
         })?;
         let meta_json = serde_json::to_string(&meta).unwrap_or_default();
+        // Whatever this write supersedes, it supersedes now — including an
+        // upload already on the wire, whose bytes describe a revision the row
+        // written just below replaces. Signalled before the enqueue so the
+        // in-flight reader stops at its next block rather than after however
+        // much of the file is left.
+        self.cancel_upload(uid);
         let superseded = if is_local_uid(uid) {
             // The node has no server-side identity to hang a revision on, so the
             // bytes ride on the create that will mint it.
@@ -2789,7 +2833,7 @@ impl Core {
                 created_at: now_millis(),
                 attempts: 0,
                 last_error: None,
-                next_attempt_at: now_millis() + DRAIN_REVISION_DEBOUNCE.as_millis() as i64,
+                next_attempt_at: now_millis() + self.revision_debounce(uid).as_millis() as i64,
             };
             let (_id, superseded) = self.db.enqueue_op(&op).map_err(|e| {
                 error!(%uid, error = %e, "queueing upload failed");
@@ -3095,6 +3139,12 @@ impl Core {
     /// worse in every way.
     fn queue_trash(&self, uid: &NodeUid, name: &str) -> Result<(), Errno> {
         self.require_uid_writable(uid)?;
+        // The bytes below are about to be discarded, and a drain worker may be
+        // reading them onto the wire right now. Stop it first: with several
+        // workers the trash op this queues can be claimed while that upload is
+        // still running, and the two would race to decide whether the file
+        // exists.
+        self.cancel_upload(uid);
         let (_, blobs) = self
             .db
             .replace_ops_with_trash(&uid.to_string(), name, now_millis())
@@ -3124,6 +3174,9 @@ impl Core {
 
     /// Drop every op queued against a node, and the staged bytes they own.
     fn discard_queued_ops(&self, uid: &NodeUid) -> Result<(), Errno> {
+        // Same reason as `queue_trash`: an upload may be reading the very blobs
+        // this is about to unlink.
+        self.cancel_upload(uid);
         let blobs = self.db.delete_ops_for_uid(&uid.to_string()).map_err(|e| {
             error!(%uid, error = %e, "dropping queued ops failed");
             Errno::EIO
@@ -6876,6 +6929,19 @@ mod tests {
         let trash = function_source(source, "fn queue_trash(");
         assert_before(trash, "replace_ops_with_trash", "discard_staged");
 
+        // §5, parallel drain: every path that unlinks a staged blob or hands
+        // the same node's work to another worker has to stop the upload that
+        // may be reading those bytes *first*. Late is a superseded revision on
+        // the wire, or two workers deciding whether the file exists.
+        for (signature, mutation) in [
+            ("fn enqueue_staged_write(", "attach_blob_to_create"),
+            ("fn queue_trash(", "replace_ops_with_trash"),
+            ("fn discard_queued_ops(", "delete_ops_for_uid"),
+        ] {
+            let function = function_source(source, signature);
+            assert_before(function, "cancel_upload", mutation);
+        }
+
         let filesystem = include_str!("filesystem.rs");
         for signature in ["fn serve_unlink(", "fn serve_rmdir("] {
             let function = function_source(filesystem, signature);

crates/pdfs-fuse/src/mount.rs

@@ -376,6 +376,8 @@ pub fn mount(
         pending: Arc::new(Mutex::new(HashMap::new())),
         hidden: Arc::new(Mutex::new(HashSet::new())),
         drain_wake: Arc::new((Mutex::new(false), Condvar::new())),
+        upload_times: Arc::new(Mutex::new(HashMap::new())),
+        upload_cancel: Arc::new(Mutex::new(HashMap::new())),
         timeline_refreshing: Arc::new(AtomicBool::new(false)),
         albums_refreshing: Arc::new(AtomicBool::new(false)),
         trash_refreshing: Arc::new(AtomicBool::new(false)),
@@ -413,11 +415,16 @@ pub fn mount(
     // And finally the staged blobs neither of those accounts for: bytes a
     // failed release left behind with nothing pointing at them.
     core.reconcile_staging();
-    {
+    // Several drain workers over one queue. One thread meant a single large
+    // upload held every queued rename, trash and small write behind it for as
+    // long as it ran; the claim column (`Db::claim_next_due_op`) keeps the
+    // workers off each other's nodes, which is the only ordering the queue
+    // actually needs. Worker 0 additionally runs the queue's idle chores.
+    for worker in 0..DRAIN_WORKERS {
         let core = core.clone();
         std::thread::Builder::new()
-            .name("pdfs-drain".into())
-            .spawn(move || core.run_pending_drain())?;
+            .name(format!("pdfs-drain-{worker}"))
+            .spawn(move || core.run_pending_drain(worker == 0))?;
     }
 
     // Start the folder-sync engine. It watches every mirror folder, polls the

crates/pdfs-fuse/src/transfers.rs

@@ -19,7 +19,7 @@ use parking_lot::Mutex;
 use std::collections::HashMap;
 use std::io::{Read, Write};
 use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::time::Instant;
 
 use pdfs_core::control::{JobItem, TransferDirection, TransferItem};
@@ -272,20 +272,50 @@ impl<R: Read> Read for OwnedCountingReader<R> {
     }
 }
 
-/// A [`Read`] that tallies bytes read through a [`TransferGuard`] (upload path).
+/// A [`Read`] that tallies bytes read through a [`TransferGuard`] (upload path),
+/// and can be told to stop.
 pub struct CountingReader<'a, R> {
     inner: R,
     guard: &'a TransferGuard,
+    cancel: Option<Arc<AtomicBool>>,
 }
 
 impl<'a, R: Read> CountingReader<'a, R> {
     pub fn new(inner: R, guard: &'a TransferGuard) -> Self {
-        Self { inner, guard }
+        Self {
+            inner,
+            guard,
+            cancel: None,
+        }
+    }
+
+    /// Abandon the upload as soon as the SDK asks for its next bytes when
+    /// `cancel` is set.
+    ///
+    /// This is the only cancellation point an upload has: the SDK takes a plain
+    /// blocking [`Read`] and pulls from it a block at a time, so refusing to
+    /// hand over the next block is what stops it. Used when a newer revision of
+    /// the same file is queued — the bytes on the wire have already been
+    /// superseded, and finishing them only spends uplink on a revision the next
+    /// op replaces.
+    pub fn with_cancel(mut self, cancel: Arc<AtomicBool>) -> Self {
+        self.cancel = Some(cancel);
+        self
     }
 }
 
 impl<R: Read> Read for CountingReader<'_, R> {
     fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        if let Some(cancel) = &self.cancel
+            && cancel.load(Ordering::Relaxed)
+        {
+            // Not `Ok(0)`: a short read is how this reader says "end of file",
+            // and the SDK would seal a truncated revision from it.
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::Interrupted,
+                "upload superseded by a newer write",
+            ));
+        }
         let n = self.inner.read(buf)?;
         self.guard.add(n as u64);
         Ok(n)