chore(docs) updated changelog

Cargo.lock

@@ -2200,6 +2200,25 @@ dependencies = [
  "system-deps",
 ]
 
+[[package]]
+name = "h2"
+version = "0.4.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155"
+dependencies = [
+ "atomic-waker",
+ "bytes",
+ "fnv",
+ "futures-core",
+ "futures-sink",
+ "http",
+ "indexmap",
+ "slab",
+ "tokio",
+ "tokio-util",
+ "tracing",
+]
+
 [[package]]
 name = "half"
 version = "2.7.1"
@@ -2367,6 +2386,7 @@ dependencies = [
  "bytes",
  "futures-channel",
  "futures-core",
+ "h2",
  "http",
  "http-body",
  "httparse",
@@ -3518,7 +3538,7 @@ checksum = "35fb2e5f958ec131621fdd531e9fc186ed768cbe395337403ae56c17a74c68ec"
 
 [[package]]
 name = "pdfs-cli"
-version = "1.7.0"
+version = "1.8.0"
 dependencies = [
  "anyhow",
  "clap 4.6.4",
@@ -3533,7 +3553,7 @@ dependencies = [
 
 [[package]]
 name = "pdfs-core"
-version = "1.7.0"
+version = "1.8.0"
 dependencies = [
  "async-trait",
  "directories",
@@ -3554,7 +3574,7 @@ dependencies = [
 
 [[package]]
 name = "pdfs-fuse"
-version = "1.7.0"
+version = "1.8.0"
 dependencies = [
  "fuser",
  "futures",
@@ -3571,7 +3591,7 @@ dependencies = [
 
 [[package]]
 name = "pdfs-gui"
-version = "1.7.0"
+version = "1.8.0"
 dependencies = [
  "anyhow",
  "async-channel",
@@ -3832,9 +3852,9 @@ dependencies = [
 
 [[package]]
 name = "proton-drive-rs"
-version = "0.5.1"
+version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3abae5e90e6889b120208159eb2c60b332612e5b708fdf3c14287307ce01715d"
+checksum = "fe0e39da5b61ceb862af2f4406870b9c1d9a1554889e8d40a32fff13882ec690"
 dependencies = [
  "async-trait",
  "base64",
@@ -3859,9 +3879,9 @@ dependencies = [
 
 [[package]]
 name = "proton-sdk"
-version = "0.5.1"
+version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3fa729310c1a66e12bb41b9f5f036c86b9857e76536556052203ffecdb0b44c3"
+checksum = "97cde8cf8e3099d1dbb56302462a632a12e0ff818c012b9c9b0a7a5200544fd6"
 dependencies = [
  "aes-gcm 0.11.0",
  "async-trait",
@@ -4252,6 +4272,7 @@ dependencies = [
  "bytes",
  "futures-core",
  "futures-util",
+ "h2",
  "http",
  "http-body",
  "http-body-util",
@@ -5110,6 +5131,7 @@ dependencies = [
  "bytes",
  "futures-core",
  "futures-sink",
+ "libc",
  "pin-project-lite",
  "tokio",
 ]

Cargo.toml

@@ -3,15 +3,15 @@ resolver = "2"
 members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"]
 
 [workspace.package]
-version = "1.7.0"
+version = "1.8.0"
 edition = "2024"
 license = "MIT"
 authors = ["Nils Pukropp <contact@narl.io>"]
 rust-version = "1.96"
 
 [workspace.dependencies]
-proton-sdk = { version = "0.5.1" }
-proton-drive-rs = { version = "0.5.1" }
+proton-sdk = { version = "0.6.0" }
+proton-drive-rs = { version = "0.6.0" }
 
 # Internal crates.
 pdfs-core = { path = "crates/pdfs-core" }

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 = 27;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -161,6 +161,46 @@ 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)?;
+            }
+        }
+        if current < 27 {
+            // Same guards as V26, for the same two reasons: the table may not
+            // exist in an old fixture, and a fixture rewound from the current
+            // schema already has the column.
+            let has_column: bool = tx.query_row(
+                "SELECT COUNT(*) FROM pragma_table_info('pending_op') \
+                 WHERE name = 'access_deferred_since'",
+                [],
+                |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_V27)?;
+            }
+        }
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -725,3 +765,46 @@ 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;
+";
+
+/// Schema v27: when an op was first deferred by the local access check.
+///
+/// An access deferral consumes no attempt and records no error — the op has not
+/// failed remotely, it has not been sent at all. That is right for a permission
+/// change the user is about to undo, and wrong for a condition that never
+/// clears: the row is re-deferred every
+/// [`DRAIN_ACCESS_RECHECK`](../../../pdfs_fuse/drain/constant.DRAIN_ACCESS_RECHECK.html)
+/// forever while `attempts`, `last_error`, `failing` and the logs all stay
+/// empty, so a queue that can never drain is indistinguishable from a busy one.
+/// Found in the field as 28 revision ops holding 18 GiB of accepted writes for
+/// 30 days, none of them visible anywhere.
+///
+/// `0` means "not currently access-deferred"; anything else is the ms at which
+/// the first consecutive deferral happened, which is what
+/// [`Db::defer_op_for_access`](super::Db::defer_op_for_access) measures the
+/// deferral against.
+const MIGRATION_V27: &str = "
+ALTER TABLE pending_op ADD COLUMN access_deferred_since INTEGER NOT NULL DEFAULT 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,
                 ),
@@ -623,11 +721,16 @@ impl Db {
 
     /// Record a failed attempt and when to next try. Leaves the row in place —
     /// the staged bytes are still the only copy of the user's write.
+    ///
+    /// Clears any access-deferral window: whatever the op was waiting for has
+    /// now been reported, so the next uncleared deferral starts its own window
+    /// rather than escalating again on the following recheck.
     pub fn record_op_failure(&self, id: i64, error: &str, next_attempt_at: i64) -> Result<()> {
         let conn = self.conn.lock();
         conn.execute(
             "UPDATE pending_op
-             SET attempts = attempts + 1, last_error = ?2, next_attempt_at = ?3
+             SET attempts = attempts + 1, last_error = ?2, next_attempt_at = ?3,
+                 access_deferred_since = 0
              WHERE id = ?1",
             params![id, error, next_attempt_at],
         )?;
@@ -653,6 +756,60 @@ impl Db {
         Ok(())
     }
 
+    /// Defer an op the local access check refused, and report how long it has
+    /// been refused for.
+    ///
+    /// Identical to [`defer_op_without_attempt`](Self::defer_op_without_attempt)
+    /// on the retry clock, plus the one thing that was missing: the first
+    /// deferral of a run stamps `access_deferred_since`, and every deferral
+    /// returns that stamp. A caller can therefore tell a deferral that is one
+    /// recheck old from one that has been repeating unchanged for days, which is
+    /// the difference between a permission the user is mid-way through changing
+    /// and a queue that will never drain.
+    ///
+    /// Returns the ms at which the current run of deferrals began — `now` on the
+    /// first one.
+    pub fn defer_op_for_access(&self, id: i64, now: i64, next_attempt_at: i64) -> Result<i64> {
+        let conn = self.conn.lock();
+        conn.execute(
+            "UPDATE pending_op
+             SET next_attempt_at = CASE
+                 WHEN next_attempt_at >= ?4 THEN next_attempt_at
+                 ELSE ?3
+             END,
+             access_deferred_since = CASE
+                 WHEN access_deferred_since = 0 THEN ?2
+                 ELSE access_deferred_since
+             END
+             WHERE id = ?1",
+            params![id, now, next_attempt_at, PARK_UNTIL],
+        )?;
+        let since = conn
+            .query_row(
+                "SELECT access_deferred_since FROM pending_op WHERE id = ?1",
+                params![id],
+                |r| r.get::<_, i64>(0),
+            )
+            .optional()?;
+        // A row that drained out from under this call leaves nothing to measure;
+        // reporting `now` says "this run just started", which is the reading that
+        // never escalates something that no longer exists.
+        Ok(since.filter(|since| *since != 0).unwrap_or(now))
+    }
+
+    /// Forget an op's access-deferral window, after an attempt the access check
+    /// let through. The op is no longer blocked, so a later deferral is the
+    /// start of a new run rather than the continuation of an old one.
+    pub fn clear_op_access_deferral(&self, id: i64) -> Result<()> {
+        let conn = self.conn.lock();
+        conn.execute(
+            "UPDATE pending_op SET access_deferred_since = 0
+             WHERE id = ?1 AND access_deferred_since <> 0",
+            params![id],
+        )?;
+        Ok(())
+    }
+
     /// Every staged blob the queue still refers to.
     ///
     /// The startup staging reconcile subtracts this from what is actually in

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

@@ -3491,6 +3491,60 @@ fn migration_v25_backfills_paths_for_an_existing_tree() {
     let _ = std::fs::remove_file(format!("{}.lock", path.display()));
 }
 
+/// A queue carried over from V26 keeps its rows, and every one of them starts
+/// out not access-deferred — a migration must not invent a deferral window that
+/// would report an ordinary queued upload as blocked.
+#[test]
+fn migration_v27_adds_the_access_deferral_column_to_a_v26_queue() {
+    let path = std::env::temp_dir().join(format!(
+        "pdfs-db-v26-fixture-{}-{}.db",
+        std::process::id(),
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos()
+    ));
+    {
+        let db = Db::open(&path).unwrap();
+        db.enqueue_op(&crate::db::PendingOp {
+            id: 0,
+            kind: crate::db::OP_REVISION.to_string(),
+            uid: "v~l".to_string(),
+            parent_uid: None,
+            name: None,
+            blob_path: Some("/tmp/blob".to_string()),
+            meta_json: Some("{}".to_string()),
+            created_at: 1,
+            attempts: 0,
+            last_error: None,
+            next_attempt_at: 0,
+        })
+        .unwrap();
+    }
+    {
+        // Put the file back in the state a released V26 database was in.
+        let conn = rusqlite::Connection::open(&path).unwrap();
+        conn.execute_batch(
+            "ALTER TABLE pending_op DROP COLUMN access_deferred_since;
+             UPDATE sync_state SET value = '26' WHERE key = 'schema_version';",
+        )
+        .unwrap();
+    }
+
+    let db = Db::open(&path).unwrap();
+    let ops = db.pending_ops().unwrap();
+    assert_eq!(ops.len(), 1, "the migration keeps the queued write");
+    let now = 5_000i64;
+    assert_eq!(
+        db.defer_op_for_access(ops[0].id, now, now + 5_000).unwrap(),
+        now,
+        "a carried-over op is not already deferred"
+    );
+
+    let _ = std::fs::remove_file(&path);
+    let _ = std::fs::remove_file(format!("{}.lock", path.display()));
+}
+
 // --- read-only connection pool (DB2) ---------------------------------------
 
 /// A read must not queue behind whatever happens to be writing.
@@ -3545,3 +3599,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,17 +40,132 @@ 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,
+    Core, DRAIN_BACKOFF_MAX, DRAIN_BACKOFF_MIN, DRAIN_IDLE_POLL, DRAIN_REVISION_DEBOUNCE,
+    DRAIN_REVISION_DEBOUNCE_MAX, ROOT_INO, UPLOAD_TIME_MEMORY, WriteAuthority, 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);
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+/// How long an op may stay access-deferred before the deferral is reported as a
+/// failure instead of retried in silence.
+///
+/// An access deferral costs no attempt and records no error, which is right for
+/// the case it was written for: a share downgraded while the queue was moving,
+/// undone a moment later. It is wrong for anything that does not clear. Until
+/// this window existed such an op was re-deferred every
+/// [`DRAIN_ACCESS_RECHECK`] indefinitely with `attempts` at zero, `last_error`
+/// null and only a `debug!` line to show for it, so `pdfs status` counted it
+/// among ordinary queued uploads and nothing ever said otherwise.
+///
+/// The condition that produced that in practice — a node absent from the local
+/// tree, which [`Db::effective_node_access`] used to report indistinguishably
+/// from a revoked permission — is now answered directly by
+/// [`Core::resolve_unknown_authority`]. This window remains the backstop for
+/// every deferral that does not resolve, including a refetch that keeps not
+/// helping.
+///
+/// Five minutes is far longer than a permission change takes to settle and far
+/// shorter than a user's patience with bytes that are not moving. Past it each
+/// recheck records a failure, so the op enters the ordinary backoff and, at
+/// [`FAILING_ATTEMPTS`], is counted and named by `Response::Status`.
+///
+/// [`Db::effective_node_access`]: pdfs_core::db::Db::effective_node_access
+/// [`FAILING_ATTEMPTS`]: pdfs_core::db::FAILING_ATTEMPTS
+const DRAIN_ACCESS_DEFER_LIMIT: Duration = Duration::from_secs(300);
+
+/// 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, PartialEq, Eq)]
 enum DrainDisposition {
     Applied,
     AccessDeferred,
+    /// An authority this op needs is missing from the local tree, so no access
+    /// answer exists for it yet. Carries the uid to ask the remote about.
+    AuthorityUnknown(NodeUid),
+}
+
+/// Why an op went back on the clock without consuming an attempt. Only affects
+/// what the user is told — every variant is deferred and reported identically.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum AccessDeferral {
+    /// An authority the tree knows about refuses the write.
+    NotWritable,
+    /// An authority missing from the tree was just re-read from the remote and
+    /// re-interned; the access check gets another look on the next pass.
+    AuthorityRefetched,
+    /// An authority missing from the tree could not be asked about.
+    RemoteUnreachable,
+}
+
+impl AccessDeferral {
+    fn log_reason(self) -> &'static str {
+        match self {
+            Self::NotWritable => "the node is not writable",
+            Self::AuthorityRefetched => "the node was missing locally and has been re-read",
+            Self::RemoteUnreachable => "the node is missing locally and the remote is unreachable",
+        }
+    }
+
+    /// Phrased to complete "<reason> for 300s." in a recorded `last_error`, so
+    /// the text the user sees names the actual condition rather than the
+    /// permission question it used to be flattened into.
+    fn failure_reason(self) -> &'static str {
+        match self {
+            Self::NotWritable => "the share has not been writable",
+            Self::AuthorityRefetched => {
+                "the node has been missing from the local tree, and re-reading it from the remote \
+                 has not restored it"
+            }
+            Self::RemoteUnreachable => {
+                "the node has been missing from the local tree and the remote has been unreachable"
+            }
+        }
+    }
 }
 
 fn pending_op_authorities(op: &PendingOp) -> Result<Vec<NodeUid>, Box<dyn std::error::Error>> {
@@ -89,7 +208,7 @@ fn pending_op_authorities(op: &PendingOp) -> Result<Vec<NodeUid>, Box<dyn std::e
 
 fn run_authorized_drain(
     op: &PendingOp,
-    mut require: impl FnMut(&NodeUid) -> Result<(), fuser::Errno>,
+    mut authority_of: impl FnMut(&NodeUid) -> WriteAuthority,
     apply: impl FnOnce() -> Result<(), Box<dyn std::error::Error>>,
 ) -> Result<DrainDisposition, Box<dyn std::error::Error>> {
     // This check is the admission/linearization point. A downgrade that lands
@@ -97,15 +216,14 @@ fn run_authorized_drain(
     // attempt. Every later queued operation rechecks. No permission lock is
     // held across remote latency.
     for authority in pending_op_authorities(op)? {
-        match require(&authority) {
-            Ok(()) => {}
-            Err(error) if error.code() == libc::EACCES => {
-                return Ok(DrainDisposition::AccessDeferred);
-            }
-            Err(error) => {
-                return Err(
-                    format!("drain access check failed with errno {}", error.code()).into(),
-                );
+        match authority_of(&authority) {
+            WriteAuthority::Writable => {}
+            WriteAuthority::Denied => return Ok(DrainDisposition::AccessDeferred),
+            // Nothing in the tree can answer for this node. Deferring on it
+            // waits for a permission change that has no reason to come, so the
+            // caller asks the remote instead (B83).
+            WriteAuthority::Unknown => {
+                return Ok(DrainDisposition::AuthorityUnknown(authority));
             }
         }
     }
@@ -127,48 +245,78 @@ 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,
-                |uid| self.require_uid_writable(uid),
+                |uid| self.uid_write_authority(uid),
                 || self.drain_op(&op),
             );
             if matches!(outcome, Ok(DrainDisposition::AccessDeferred)) {
-                let next_attempt_at = now_millis() + DRAIN_ACCESS_RECHECK.as_millis() as i64;
-                if let Err(error) = self.db.defer_op_without_attempt(op.id, next_attempt_at) {
-                    error!(
-                        uid = %op.uid,
-                        %error,
-                        "deferring access-blocked pending operation failed"
-                    );
-                    self.wait_for_drain_work();
-                } else {
-                    debug!(
-                        uid = %op.uid,
-                        next_attempt_at,
-                        "pending operation deferred until access is writable"
-                    );
-                }
+                self.defer_for_access(&op, AccessDeferral::NotWritable);
                 continue;
             }
+            if let Ok(DrainDisposition::AuthorityUnknown(ref authority)) = outcome {
+                self.resolve_unknown_authority(&op, authority);
+                continue;
+            }
+            // The access check let this attempt through, so any earlier run of
+            // deferrals is over and must not be counted against the next one.
+            if matches!(outcome, Ok(DrainDisposition::Applied))
+                && let Err(error) = self.db.clear_op_access_deferral(op.id)
+            {
+                debug!(uid = %op.uid, %error, "clearing an access-deferral window failed");
+            }
             if let Err(e) = outcome {
                 let attempts = op.attempts + 1;
                 let err_str = e.to_string();
@@ -188,9 +336,234 @@ 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);
+        }
+    }
+
+    /// Put an op the access check refused back on the clock, and decide whether
+    /// its refusal is still news.
+    ///
+    /// Inside [`DRAIN_ACCESS_DEFER_LIMIT`] a deferral is what it always was: no
+    /// attempt consumed, no error recorded, retried in five seconds. Past it the
+    /// deferral is reported through [`Db::record_op_failure`] like any other
+    /// stuck op, which is what puts it in `attempts`, in `last_error`, and — at
+    /// [`FAILING_ATTEMPTS`] — in the `failing_ops` count and error text that
+    /// `pdfs status` and the GUI show. The row and its staged blob stay exactly
+    /// where they are either way; this changes only whether the user can see
+    /// them.
+    ///
+    /// [`Db::record_op_failure`]: pdfs_core::db::Db::record_op_failure
+    /// [`FAILING_ATTEMPTS`]: pdfs_core::db::FAILING_ATTEMPTS
+    fn defer_for_access(&self, op: &PendingOp, reason: AccessDeferral) {
+        let now = now_millis();
+        let next_attempt_at = now + DRAIN_ACCESS_RECHECK.as_millis() as i64;
+        let since = match self.db.defer_op_for_access(op.id, now, next_attempt_at) {
+            Ok(since) => since,
+            Err(error) => {
+                error!(
+                    uid = %op.uid,
+                    %error,
+                    "deferring access-blocked pending operation failed"
+                );
+                self.release_claim(op);
+                self.wait_for_drain_work();
+                return;
+            }
+        };
+        let blocked = Duration::from_millis(now.saturating_sub(since).max(0) as u64);
+        if since == now {
+            // The first deferral of a run, at `warn!`: a write the mount accepted
+            // is not going anywhere for now, and that is worth a line even when
+            // the next recheck clears it.
+            warn!(
+                uid = %op.uid,
+                kind = %op.kind,
+                reason = reason.log_reason(),
+                "pending operation deferred"
+            );
+        }
+        if blocked >= DRAIN_ACCESS_DEFER_LIMIT {
+            let attempts = op.attempts + 1;
+            let backoff = DRAIN_BACKOFF_MIN
+                .saturating_mul(1u32 << attempts.min(6))
+                .min(DRAIN_BACKOFF_MAX);
+            let error = format!(
+                "{} for {}s. The staged bytes are kept.",
+                reason.failure_reason(),
+                blocked.as_secs()
+            );
+            warn!(uid = %op.uid, attempts, blocked_secs = blocked.as_secs(),
+                  "pending operation has been access-blocked past the limit; reporting it");
+            if let Err(error) =
+                self.db
+                    .record_op_failure(op.id, &error, now_millis() + backoff.as_millis() as i64)
+            {
+                error!(uid = %op.uid, %error, "recording an access-deferral failure failed");
+            }
+        } else {
+            debug!(
+                uid = %op.uid,
+                next_attempt_at,
+                blocked_secs = blocked.as_secs(),
+                reason = reason.log_reason(),
+                "pending operation deferred until access is writable"
+            );
+        }
+        self.release_claim(op);
+    }
+
+    /// Answer an authority the local tree has no row for by asking the remote,
+    /// rather than deferring on a permission question nobody asked (B83).
+    ///
+    /// Three outcomes, and the point of the whole exercise is that they are
+    /// three:
+    ///
+    /// * the node is still there — the tree simply lost it (a pruned subtree, a
+    ///   mount whose registration went away). Re-intern it and the next pass has
+    ///   a real access answer to work with.
+    /// * the node is gone remotely. That is a permanent, *known* condition, so
+    ///   it is recorded as a failure immediately instead of waiting out
+    ///   [`DRAIN_ACCESS_DEFER_LIMIT`] to say so. The row and its staged blob are
+    ///   kept, as on every other failure path — the bytes are still the user's,
+    ///   and `pdfs recover` is how they get them back.
+    /// * we could not ask. Ordinary deferral; the network is the network.
+    fn resolve_unknown_authority(&self, op: &PendingOp, authority: &NodeUid) {
+        match self.fetch_node_remote(authority) {
+            Ok(Some(node)) => {
+                if let Err(error) = self.db.upsert_node(&node) {
+                    error!(uid = %op.uid, %authority, %error,
+                           "re-interning a queued operation's missing authority failed");
+                } else {
+                    info!(uid = %op.uid, %authority,
+                          "re-interned an authority missing from the local tree; retrying");
+                }
+                // Deferred, not applied: the re-read has to go through the same
+                // access check as everything else, on the next pass. The window
+                // keeps running, so a refetch that never actually helps still
+                // gets reported instead of looping in silence.
+                self.defer_for_access(op, AccessDeferral::AuthorityRefetched);
+            }
+            Ok(None) => {
+                let attempts = op.attempts + 1;
+                let backoff = DRAIN_BACKOFF_MIN
+                    .saturating_mul(1u32 << attempts.min(6))
+                    .min(DRAIN_BACKOFF_MAX);
+                let error = format!(
+                    "{authority} no longer exists remotely, so this operation has nothing to \
+                     apply to. The staged bytes are kept."
+                );
+                warn!(uid = %op.uid, %authority, attempts,
+                      "a queued operation's authority is gone remotely; reporting it");
+                if let Err(error) = self.db.record_op_failure(
+                    op.id,
+                    &error,
+                    now_millis() + backoff.as_millis() as i64,
+                ) {
+                    error!(uid = %op.uid, %error, "recording a missing-authority failure failed");
+                }
+                self.release_claim(op);
+            }
+            Err(error) => {
+                debug!(uid = %op.uid, %authority, %error,
+                       "could not ask the remote about a missing authority");
+                self.defer_for_access(op, AccessDeferral::RemoteUnreachable);
+            }
+        }
+    }
+
+    /// 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 +1280,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
@@ -1227,12 +1622,10 @@ mod tests {
         let applied = Cell::new(false);
         let disposition = run_authorized_drain(
             &op,
-            |uid| {
-                db.effective_node_access(uid)
-                    .unwrap()
-                    .filter(|access| access.writable())
-                    .map(|_| ())
-                    .ok_or(fuser::Errno::EACCES)
+            |uid| match db.effective_node_access(uid).unwrap() {
+                Some(access) if access.writable() => WriteAuthority::Writable,
+                Some(_) => WriteAuthority::Denied,
+                None => WriteAuthority::Unknown,
             },
             || {
                 applied.set(true);
@@ -1272,7 +1665,7 @@ mod tests {
 
         let disposition = run_authorized_drain(
             &op,
-            |_| Err(fuser::Errno::EACCES),
+            |_| WriteAuthority::Denied,
             || {
                 applied.set(true);
                 Ok(())
@@ -1293,13 +1686,157 @@ mod tests {
         let _ = std::fs::remove_dir_all(dir);
     }
 
+    /// A deferral that never clears has to stop being silent. Found in the field
+    /// as revision ops whose node had left the local tree: `require_uid_access`
+    /// reads an unknown node as `EACCES`, so they were re-deferred every five
+    /// seconds for thirty days with `attempts` at zero, `last_error` null and
+    /// `failing_ops` at zero — 18 GiB of accepted writes that `pdfs status`
+    /// reported as ordinary queued uploads.
+    #[test]
+    fn an_access_deferral_that_outlives_the_limit_is_reported_as_failing() {
+        use std::sync::atomic::{AtomicU64, Ordering};
+
+        static N: AtomicU64 = AtomicU64::new(0);
+        let dir = std::env::temp_dir().join(format!(
+            "pdfs-drain-defer-limit-{}-{}",
+            std::process::id(),
+            N.fetch_add(1, Ordering::Relaxed)
+        ));
+        std::fs::create_dir_all(&dir).unwrap();
+        let db = Db::open(&dir.join("cache.db")).unwrap();
+        db.enqueue_op(&pending(OP_REVISION)).unwrap();
+        let op = db.pending_ops().unwrap().remove(0);
+
+        // Two deferrals a recheck apart: the window opens on the first and the
+        // second is measured against it, not against itself.
+        let now = 1_000_000i64;
+        let since = db.defer_op_for_access(op.id, now, now + 5_000).unwrap();
+        assert_eq!(since, now, "the first deferral opens the window");
+        let later = now + DRAIN_ACCESS_RECHECK.as_millis() as i64;
+        assert_eq!(
+            db.defer_op_for_access(op.id, later, later + 5_000).unwrap(),
+            now,
+            "a later deferral keeps the original stamp"
+        );
+        let queued = db.pending_ops().unwrap();
+        assert_eq!(queued[0].attempts, 0, "deferrals consume no attempt");
+        assert!(queued[0].last_error.is_none());
+
+        // Past the limit the same deferral is reported like any other stuck op.
+        let past = now + DRAIN_ACCESS_DEFER_LIMIT.as_millis() as i64;
+        assert_eq!(
+            db.defer_op_for_access(op.id, past, past + 5_000).unwrap(),
+            now
+        );
+        db.record_op_failure(op.id, "not writable locally", past + 10_000)
+            .unwrap();
+        let failed = db.pending_ops().unwrap();
+        assert_eq!(failed.len(), 1, "the row and its staged bytes stay queued");
+        assert_eq!(failed[0].attempts, 1);
+        assert_eq!(
+            failed[0].last_error.as_deref(),
+            Some("not writable locally")
+        );
+
+        // Reporting closes the window, so the next deferral starts a fresh one
+        // rather than escalating on every recheck from here on.
+        let after = past + 20_000;
+        assert_eq!(
+            db.defer_op_for_access(op.id, after, after + 5_000).unwrap(),
+            after,
+            "recording the failure re-armed the window"
+        );
+        drop(db);
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
+    /// An op that gets through the access check is no longer blocked, so a
+    /// deferral after it must not inherit the earlier run's age and escalate
+    /// immediately.
+    #[test]
+    fn clearing_the_window_makes_the_next_deferral_a_fresh_run() {
+        use std::sync::atomic::{AtomicU64, Ordering};
+
+        static N: AtomicU64 = AtomicU64::new(0);
+        let dir = std::env::temp_dir().join(format!(
+            "pdfs-drain-defer-clear-{}-{}",
+            std::process::id(),
+            N.fetch_add(1, Ordering::Relaxed)
+        ));
+        std::fs::create_dir_all(&dir).unwrap();
+        let db = Db::open(&dir.join("cache.db")).unwrap();
+        db.enqueue_op(&pending(OP_REVISION)).unwrap();
+        let op = db.pending_ops().unwrap().remove(0);
+
+        let now = 2_000_000i64;
+        assert_eq!(
+            db.defer_op_for_access(op.id, now, now + 5_000).unwrap(),
+            now
+        );
+        db.clear_op_access_deferral(op.id).unwrap();
+        let later = now + DRAIN_ACCESS_DEFER_LIMIT.as_millis() as i64;
+        assert_eq!(
+            db.defer_op_for_access(op.id, later, later + 5_000).unwrap(),
+            later,
+            "the window reopens at the new deferral, not the old one"
+        );
+        drop(db);
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
+    #[test]
+    fn an_authority_the_tree_does_not_have_is_reported_as_unknown_not_deferred() {
+        // The B83 distinction, at the point it is made: a uid with no row must
+        // not come back as `AccessDeferred`, because deferring waits on a
+        // permission change that nothing is going to make. The caller answers
+        // an `AuthorityUnknown` by asking the remote instead.
+        let op = pending(OP_REVISION);
+        let applied = Cell::new(false);
+        let disposition = run_authorized_drain(
+            &op,
+            |_| WriteAuthority::Unknown,
+            || {
+                applied.set(true);
+                Ok(())
+            },
+        )
+        .unwrap();
+        assert_eq!(
+            disposition,
+            DrainDisposition::AuthorityUnknown(parse_node_uid(&op.uid).unwrap())
+        );
+        assert!(!applied.get(), "an unresolved authority never applies");
+    }
+
+    #[test]
+    fn a_missing_parent_authority_names_the_parent_not_the_node() {
+        // A create's authority is its parent, so that is the uid the remote has
+        // to be asked about — reporting the node instead would refetch the
+        // thing that does not exist yet.
+        let mut op = pending(OP_CREATE);
+        let parent = folder_node("vanished-parent", None);
+        op.parent_uid = Some(parent.uid.to_string());
+        let disposition =
+            run_authorized_drain(&op, |_| WriteAuthority::Unknown, || Ok(())).unwrap();
+        assert_eq!(
+            disposition,
+            DrainDisposition::AuthorityUnknown(parent.uid.clone())
+        );
+    }
+
     #[test]
     fn owner_and_editor_drain_authorities_apply_normally() {
         for access in [Access::Owner, Access::Editor] {
             let applied = Cell::new(false);
             let disposition = run_authorized_drain(
                 &pending(OP_REVISION),
-                |_| access.writable().then_some(()).ok_or(fuser::Errno::EACCES),
+                |_| {
+                    if access.writable() {
+                        WriteAuthority::Writable
+                    } else {
+                        WriteAuthority::Denied
+                    }
+                },
                 || {
                     applied.set(true);
                     Ok(())
@@ -1413,3 +1950,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.
@@ -935,6 +973,27 @@ impl Core {
     /// downgrade cannot be hidden by stale live state, and a live downgrade
     /// cannot be hidden by a restored DB row. Missing UIDs fail closed.
     fn require_uid_writable(&self, uid: &NodeUid) -> Result<(), Errno> {
+        match self.uid_write_authority(uid) {
+            WriteAuthority::Writable => Ok(()),
+            // A uid nobody can speak for is refused here just as a denied one
+            // is: a syscall is a live request against a live tree, and a stale
+            // handle must not be admitted because the tree forgot the node.
+            // The drain, which replays intent recorded long ago, is the one
+            // caller that has to tell the two apart — see
+            // [`Core::uid_write_authority`].
+            WriteAuthority::Denied | WriteAuthority::Unknown => Err(Errno::EACCES),
+        }
+    }
+
+    /// The full answer behind [`Core::require_uid_writable`]: whether every
+    /// authority agrees this node is writable, refuses it, or has never heard
+    /// of it.
+    ///
+    /// The third case is the one worth naming. A node absent from the local
+    /// tree yields no access row, and collapsing that into "denied" told the
+    /// drain to wait for a permission change that was never coming, because
+    /// nothing about the node's permissions was ever the problem (B83).
+    pub(crate) fn uid_write_authority(&self, uid: &NodeUid) -> WriteAuthority {
         let mut live_access = Vec::new();
         {
             let state = self.state();
@@ -950,22 +1009,38 @@ impl Core {
                 live_access.push(access);
             }
         }
-        require_uid_access(&self.db, uid, &live_access)
+        uid_write_authority(&self.db, uid, &live_access)
     }
 }
 
-fn require_uid_access(db: &Db, uid: &NodeUid, live_access: &[Access]) -> Result<(), Errno> {
+/// Whether a node may be written, as agreed by every authority that has an
+/// opinion — or the fact that none of them does.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum WriteAuthority {
+    /// The persisted tree and every resident mount agree: writable.
+    Writable,
+    /// At least one authority refuses it, or could not be consulted. Fails
+    /// closed, so a lookup error lands here.
+    Denied,
+    /// The node is not in the local tree, so there is no authority to consult.
+    /// Not the same as a refusal, and callers that can ask the remote should.
+    Unknown,
+}
+
+fn uid_write_authority(db: &Db, uid: &NodeUid, live_access: &[Access]) -> WriteAuthority {
     let persisted = match db.effective_node_access(uid) {
         Ok(Some(access)) => access,
-        Ok(None) => return Err(Errno::EACCES),
+        Ok(None) => return WriteAuthority::Unknown,
         Err(error) => {
             error!(%uid, %error, "db effective access lookup failed");
-            return Err(Errno::EACCES);
+            return WriteAuthority::Denied;
         }
     };
-    (persisted.writable() && live_access.iter().all(|access| access.writable()))
-        .then_some(())
-        .ok_or(Errno::EACCES)
+    if persisted.writable() && live_access.iter().all(|access| access.writable()) {
+        WriteAuthority::Writable
+    } else {
+        WriteAuthority::Denied
+    }
 }
 
 /// One kernel notification, recorded while the State lock is held so it can be
@@ -2756,6 +2831,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 +2870,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 +3176,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 +3211,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
@@ -5701,9 +5791,19 @@ mod tests {
         preserve_on_access_denied, publish_virtual_root_in_listing,
         reconcile_virtual_root_in_listing, release_can_discard_unlinked,
         release_must_retain_queued_trash, release_unlinked_entry, rename_needs_queue,
-        require_node_parent_access, require_rename_access, require_uid_access,
-        resolve_anywhere_with, shared_with_me_uid, take_self_change, virtual_node,
+        require_node_parent_access, require_rename_access, resolve_anywhere_with,
+        shared_with_me_uid, take_self_change, uid_write_authority, virtual_node,
     };
+    use super::{Db, WriteAuthority};
+
+    /// The collapse [`Core::require_uid_writable`] performs, without a `Core`:
+    /// every non-writable authority is one `EACCES` on the syscall path.
+    fn require_uid_access(db: &Db, uid: &NodeUid, live: &[Access]) -> Result<(), Errno> {
+        match uid_write_authority(db, uid, live) {
+            WriteAuthority::Writable => Ok(()),
+            WriteAuthority::Denied | WriteAuthority::Unknown => Err(Errno::EACCES),
+        }
+    }
     use crate::filesystem::access_allowed;
     use pdfs_core::cache::{Baseline, StagedWrite};
     use pdfs_core::db::{OP_REVISION, OP_TRASH, PendingOp};
@@ -6710,6 +6810,19 @@ mod tests {
             require_uid_access(&db, &stale, &[]).unwrap_err().code(),
             libc::EACCES
         );
+        // Same refusal on the syscall path, but a distinguishable one
+        // underneath: the drain asks the remote about `Unknown` rather than
+        // waiting on a permission change (B83).
+        assert_eq!(
+            uid_write_authority(&db, &stale, &[]),
+            WriteAuthority::Unknown,
+            "a uid the tree has no row for is unknown, not denied"
+        );
+        assert_eq!(
+            uid_write_authority(&db, &uid, &[]),
+            WriteAuthority::Denied,
+            "a uid the tree knows and refuses is denied"
+        );
         assert_eq!(db.pending_ops().unwrap().len(), before);
     }
 
@@ -6876,6 +6989,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)

docs/ARCHITECTURE.md

@@ -153,8 +153,9 @@ Because Proton Drive does not support partial byte writes, modified files must b
 2. **Close/Release (`queue_revision`):** When the application closes the file descriptor, the daemon:
    - Fetches any untouched gaps from the remote base file to compile the full file.
    - Durably publishes the scratch data and authored-range sidecar into `staging` under a `{uid}-{millis}-{counter}` name. Temporary data and metadata are synced before atomic rename, and their directory is synced before the source is removed.
-   - Transactionally queues or supersedes a pending database operation (`PendingOp`), so insertion failure cannot erase the previously acknowledged upload. For `OP_REVISION` ops, execution is debounced by 2 seconds (`DRAIN_REVISION_DEBOUNCE = 2s`) to give rapid follow-up writes (e.g. `aria2c` preallocation followed by writing) time to supersede the staged blob before network transmission.
-3. **Async Drain Thread (`run_pending_drain`):** The background drain worker picks up the database operations queue, handles revision uploads, resolves conflicts, and cleans up staging files. Upon landing a revision upload (`refresh_after_upload`), it rebaselines both still-queued ops (`rebaseline_pending`) and open write handles targeting the same node to prevent false self-conflict copies on subsequent writes.
+   - Transactionally queues or supersedes a pending database operation (`PendingOp`), so insertion failure cannot erase the previously acknowledged upload. For `OP_REVISION` ops, execution is debounced to give rapid follow-up writes (e.g. `aria2c` preallocation followed by writing) time to supersede the staged blob before network transmission. The debounce is adaptive (`Core::revision_debounce`): it starts at `DRAIN_REVISION_DEBOUNCE = 2s` and, once a node has been uploaded, widens toward how long that upload actually took, bounded by `DRAIN_REVISION_DEBOUNCE_MAX = 60s`. A file saved faster than it can be sent therefore supersedes in the queue rather than mid-upload.
+   - Signals `Core::cancel_upload` for the node before touching the queue. A drain worker may already be reading the blob this write supersedes; the flag is read by the upload's `CountingReader`, which refuses the SDK's next block. `queue_trash` and `discard_queued_ops` do the same, because they unlink the blob outright.
+3. **Async Drain Threads (`run_pending_drain`):** `DRAIN_WORKERS = 3` background workers share the operations queue. They pick work up through `Db::claim_next_due_op`, which marks the row `claimed_at` in the same transaction that selects it and excludes any op whose uid another worker already holds — ordering only has to hold *per node*, and that exclusion is what guarantees it. Worker 0 additionally runs the queue's idle chores (LRU touch flush, `recover_fsynced_writes`). A claim is process-local state: the single-writer `flock` means one found at open belongs to a crashed run, and `Db::clear_op_claims` drops the lot so those ops are not invisible forever. Each worker handles revision uploads, resolves conflicts, and cleans up staging files. Upon landing a revision upload (`refresh_after_upload`), it rebaselines both still-queued ops (`rebaseline_pending`) and open write handles targeting the same node to prevent false self-conflict copies on subsequent writes.
 
 ```mermaid
 sequenceDiagram

docs/BUGS.md

@@ -12,6 +12,82 @@ Conventions:
 
 ---
 
+## B83 — An access-deferred op retries forever and is reported as an ordinary queued upload
+
+**Status:** Fixed (partly verified) 2026-08-16 — both halves are in and
+unit-tested. The reporting half was confirmed against the live 28-op backlog on
+1.8.0 (schema migrated to v27, all 28 rows window-stamped, `WARN` lines per op);
+the resolution half has not yet been observed draining that backlog.
+**Found:** 2026-08-16, running `scripts/fuse-acceptance.sh --live` on a
+throwaway CLI-made mount. Four regression cases (B69, B70, B74, B79) failed with
+`TimeoutError: daemon mutation queue did not drain` — the suite's
+`wait_for_queue()` polls the *global* `status.mount.pending_uploads`, which had
+been sitting at 28 for 30 days on the development machine. `pdfs status` called
+them 28 queued uploads holding 18.4 GiB; `pdfs transfers` showed nothing;
+`failing_ops` and `parked_uploads` were both 0.
+**Where:** `run_pending_drain` / `run_authorized_drain`
+(`crates/pdfs-fuse/src/drain.rs`), `require_uid_writable`
+(`crates/pdfs-fuse/src/lib.rs`), `Db::effective_node_access`
+(`crates/pdfs-core/src/db/share_access.rs`),
+`Db::defer_op_without_attempt` (`crates/pdfs-core/src/db/ops.rs`).
+
+**Root cause.** `effective_node_access` returns `Ok(None)` when the uid has no
+row in `nodes`, and `require_uid_access` maps that to `EACCES` — "I have never
+heard of this node" and "you may not write this node" are the same answer. The
+drain reads `EACCES` as `DrainDisposition::AccessDeferred` and calls
+`defer_op_without_attempt`, which by design touches neither `attempts` nor
+`last_error`. So the op was re-deferred every `DRAIN_ACCESS_RECHECK` (5s)
+indefinitely, and every channel that could have surfaced it stayed empty:
+`attempts` 0, `last_error` NULL, `parked` counts only `PARK_UNTIL` rows,
+`failing` counts only `attempts >= FAILING_ATTEMPTS`, and the one log line was
+`debug!`. On the machine that found it, all 28 uids were absent from `nodes`:
+
+```sql
+select (select count(*) from pending_op p
+          where exists(select 1 from nodes n where n.uid = p.uid)) as in_nodes,
+       (select count(*) from pending_op) as total;
+-- 0 | 28
+```
+
+**Fix, part 1 — stop guessing.** `WriteAuthority { Writable, Denied, Unknown }`
+replaces the `Result<(), Errno>` the drain's access check used to return
+(`Core::uid_write_authority`). `require_uid_writable` still collapses both
+refusals to `EACCES`, which is right for a syscall: a stale handle must not be
+admitted because the tree forgot the node. The drain does not collapse them —
+`Unknown` becomes `DrainDisposition::AuthorityUnknown(uid)`, and
+`Core::resolve_unknown_authority` asks the remote:
+
+- node still there → `upsert_node` re-interns it and the op is deferred for one
+  recheck, so the next pass gets a real access answer;
+- node gone (`Ok(None)` / `is_gone`) → `record_op_failure` **immediately**, with
+  a message naming the vanished uid, rather than waiting out the 5-minute window
+  to say something vaguer;
+- could not ask → ordinary deferral.
+
+The authority asked about is the op's own (`pending_op_authorities`), so a
+`create`/`mkdir` refetches its *parent*, not the node it has not made yet. The
+row and its staged blob survive every branch.
+
+**Fix, part 2 — say something.** Schema v27 adds `pending_op.access_deferred_since`.
+`Db::defer_op_for_access` stamps it on the first deferral of a run and returns
+the stamp on every one; `Core::defer_for_access` warns on the first deferral and,
+once the run passes `DRAIN_ACCESS_DEFER_LIMIT` (5 min), reports it through
+`record_op_failure` so it enters the ordinary backoff and shows up in
+`failing_ops` and the status error text. `record_op_failure` clears the window so
+each escalation re-arms it rather than firing every recheck, and an attempt the
+access check admits clears it too (`clear_op_access_deferral`). The row and its
+staged blob are untouched on every path — this changes only what the user can
+see.
+
+The deferral window stays as the backstop for everything that does not resolve,
+including a refetch that keeps not helping, and `AccessDeferral` now carries the
+reason into both the log line and the recorded `last_error`.
+
+**Still open:** the acceptance runner waits on the *global* queue
+(`ManagedSyncPair.wait_for_queue`), so any unrelated stuck op fails those four
+cases on a machine that has one — that is what turned this backlog into four
+apparent FUSE regressions.
+
 ## B82 — A `parent_uid` cycle hangs the daemon on any short search
 
 **Status:** Fixed (unverified) 2026-08-12 — unit-tested, not driven against a

docs/CHANGELOG.md

@@ -0,0 +1,292 @@
+# Changelog
+
+All notable changes to the Linux client since the 1.0.0 release. Format follows
+[Keep a Changelog](https://keepachangelog.com/1.1.0/); versions follow semver, and every
+release is tagged `vX.Y.Z` with `Cargo.toml` and `packaging/PKGBUILD` in agreement.
+
+`Schema:` lines record the SQLite `SCHEMA_VERSION` (`pdfs-core/src/db/migrations.rs`) a
+release ships. Migrations are forward-only — a database written by a newer client is a hard
+refuse-to-open, not a downgrade, so rolling back a release means restoring the cache from
+scratch (user data in `staging/` and `recovery/` is never touched by this).
+
+## [1.8.0] — 2026-08-16
+
+A queued write whose node had left the local tree was retried every five seconds forever and
+counted as an ordinary pending upload — 28 of them, holding 18.4 GiB, had been doing that for
+30 days on the machine that found it (`docs/BUGS.md` B83). Schema: 27 — migration V27 adds
+`pending_op.access_deferred_since`.
+
+### Fixed
+- The drain no longer reads "I have never heard of this node" as "you may not write this
+  node". A missing authority is now resolved against the remote: the node is re-interned if
+  it is still there, and reported as a real failure if it is gone. Ops that had been stuck
+  invisibly for weeks now say what is wrong with them on the first pass.
+- A create or mkdir resolves its *parent* as the authority, so the refetch asks about the
+  folder rather than the node it has not made yet.
+
+### Added
+- Access deferrals are bounded. A deferral still costs no attempt and records no error for
+  the first five minutes — the case it exists for is a share downgraded mid-queue and undone
+  a moment later — but past that window each recheck records a failure, so the op enters the
+  ordinary backoff and appears in `failing_ops` and the `pdfs status` error text like any
+  other stuck operation. The first deferral of a run is logged at `warn` rather than `debug`.
+
+### Notes
+- No staged data is touched by any of this. A reported failure keeps its `pending_op` row and
+  its staged blob exactly as before; the change is in what the user is told, and in the queue
+  no longer waiting on a permission change that was never coming.
+
+## [1.7.1] — 2026-08-16
+
+Write-back throughput follow-up to 1.7.0, same audit
+(`docs/plans/audit-2026-08-15-perf-locks.md`). Schema: 26 — migration V26 adds
+`pending_op.claimed_at`.
+
+### Added
+- Parallel write-back drain: three workers share the queue through a `pending_op.claimed_at`
+  claim column (migration V26), so a large upload no longer holds every queued rename, trash
+  and small write behind it. Ordering is preserved per node — the claim query never offers
+  two workers ops for the same uid — and claims left by a crashed run are cleared at open.
+- An upload already on the wire is cancelled when a newer write, a trash or a discard
+  supersedes it, instead of spending the uplink on a revision the next operation replaces.
+
+### Changed
+- The revision debounce is adaptive: it widens toward how long that node's last upload
+  actually took (bounded to 2–60 s), so a file saved faster than it can be sent supersedes
+  in the queue rather than mid-upload.
+- Workspace deps bumped to `proton-sdk` / `proton-drive-rs` 0.6.0.
+
+## [1.7.0] — 2026-08-16
+
+Performance and robustness pass over the mount, from the architecture audit in
+`docs/plans/audit-2026-08-15-perf-locks.md`. Schema: 25.
+
+### Added
+- Sequential-read prefetch (`Core::prefetch_ahead`): per-file sequential detection, depth
+  doubling 2→8, reset to 0 after a seek, skipping blocks already cached, and bounded by its
+  own permit budget so it can never queue ahead of a demand read.
+- Single-flight block fetching (`BlockFlight`) keyed on `(uid, mtime, size, block)`, so a
+  demand read, kernel read-ahead and prefetch of the same block share one download+decrypt.
+- In-memory block LRU (`BlockRing`) in front of *both* block paths; a block read off disk is
+  promoted into it, so the following ~31 kernel reads of that 4 MiB block cost nothing.
+- Read-only SQLite connection pool (`ReadPool` / `Db::read`) with every `SELECT`-only method
+  routed at it, so a read no longer waits on the write connection.
+- `nodes.path` persisted and backfilled, replacing the recursive `path_of` walk.
+- Single-writer `flock` on the database; a `SQLITE_CORRUPT`/`NOTADB` file is quarantined as
+  `cache.db.corrupt-<ts>` (WAL sidecars too) and rebuilt rather than failing the mount.
+- `VACUUM` and the deep `integrity_check` now ack immediately and run as background jobs;
+  `pdfs doctor` polls for the verdict.
+- Staging reconcile at mount: staged blobs no queued operation refers to are re-queued, or
+  named in the log when they cannot be addressed.
+- Queue/staging visibility — `Status` gained `parked_uploads`, `failing_ops`,
+  `failing_error`, `staged_bytes`, `staged_oldest_secs`; `pdfs status` prints them when
+  non-zero.
+- `statfs` implemented against the account quota; FUSE `init` negotiates `max_readahead`,
+  `max_background`, `PARALLEL_DIROPS` and `CACHE_SYMLINKS`; `blksize` raised to 1 MiB.
+
+### Changed
+- `State` no longer writes to SQLite under the inode lock — mutations queue into an outbox
+  applied after the lock is released, so interning a 5000-child listing no longer commits a
+  transaction with the whole mount frozen.
+- Re-listing an unchanged folder no longer touches its subtree; when a subtree really moved,
+  one recursive walk carries both the new path and the inherited trashed flag.
+- `release(2)` no longer gap-fills from the network: it stages an incomplete blob from local
+  caches only and the drain thread finishes it off the dispatch loop.
+- A partial write over an undrained one is merged instead of refused.
+- Size-upgrade waiters park on a queue instead of occupying a worker thread, capped at 8
+  folders in flight; past the cap `stat` answers provisionally.
+- Block-cache disk writes moved off the read's critical path; no more fsync per cached block.
+- Cache LRU touches buffered in memory and flushed in one transaction.
+- Search: sync-folder list and pin lookups hoisted out of the per-hit loop.
+- SQLite tuned (`cache_size`, `mmap_size`, `temp_store=MEMORY`, `wal_autocheckpoint`) and
+  three new indexes added.
+- Mount registry lock released before FUSE teardown/join; kernel notifications batched under
+  the state lock and flushed after it.
+
+### Fixed
+- **Data loss:** `ENOSPC` is now propagated from `write`/`fsync`/staging instead of being
+  swallowed, with emergency eviction wired up and scratch marked durable before any staging
+  error; a failed `record_pending_write` fails `release` rather than dropping the write.
+- **Data loss:** the conflict sweep no longer discards queued operations after the trash
+  round trip.
+- **Data loss:** `rescue_scratch` moves a blob with an unreadable sidecar to `recovery/`
+  unaccompanied instead of fabricating a sidecar for it.
+- Own-sealed revisions are persisted (7-day TTL), so a restart no longer reopens the
+  transient-download fork window fixed in 1.1.0.
+- Depth-capped `is_pinned` (cycle-safe), daemon write timeout, and every client-supplied
+  `limit` clamped.
+- `recover_fsynced_writes` no longer scans the recovery directory on every idle pass.
+
+## [1.6.0] — 2026-08-12
+
+Search-index correctness and ranking (`docs/plans/search-index.md`). Schema: 21 —
+migration V21 rebuilds `nodes_fts`.
+
+### Fixed
+- **B80** — 29% of the account was missing from the search index. `node_is_indexable_tx`
+  required a null-parent ancestor, which only the My Files root has, so every node under a
+  device folder was dropped from `nodes_fts` on its first upsert after the v16 backfill
+  (2,705 of 9,252 nodes on the account this was diagnosed against). The rule now matches the
+  backfill, migration V21 replays it, nodes under a trashed folder stay out, and the walk is
+  depth-capped.
+- **B81** — the prompt labelled every Drive hit "My files", including device folders; pin
+  rows also took `is_dir` from `Pin::recursive`.
+- **B82** — a `parent_uid` cycle hung the daemon on any short search: the recursive CTEs in
+  `path_of` and the pin walk use `UNION ALL`, which does not deduplicate. `path_of` is now
+  depth-capped at 256 and returns the truncated path.
+
+### Changed
+- Local search ranking: whole-query prefix matches get their own lane (newest first) ahead
+  of trigram rank, which on a home directory full of dependency trees was burying the
+  obvious answer — 73% of the 500 best-ranked candidates for "test" came from one Go module
+  cache. The single-character lane, which recovers a typo that destroys every trigram, stays
+  behind it.
+
+## [1.5.0] — 2026-08-11
+
+Schema: 20.
+
+### Added
+- **`pdfs-prompt --dmenu`** — the same search through an external dmenu-style launcher
+  (fuzzel, rofi, wofi, tofi, bemenu, dmenu) instead of the built-in GTK HUD, so the prompt
+  matches a tiling-WM desktop. A launcher filters a fixed list rather than re-querying per
+  keystroke, so searching is a two-step loop; `--query` skips the first step, fuzzel and rofi
+  get file-type icons. Configurable as `"prompt": { "mode": "dmenu", "menu": [...] }`, with
+  `--gtk` overriding for one invocation.
+- **Configurable open-with rules** (`pdfs-core/src/opener.rs`): an ordered list of name/class
+  patterns in `config.json` under `open_with`, each mapped to argv and optionally wrapped in
+  a terminal, first match wins, falling back to `xdg-open`. Lets "open a text file" mean
+  `nvim` in a terminal without breaking every graphical caller. Absent config reproduces the
+  old behaviour exactly.
+
+### Fixed
+- The prompt and dmenu openers hand applications the cached file while matching rules
+  against the *original* Drive filename, so e.g. `.md` files stop being routed by their
+  cache-path extension.
+
+## [1.4.0] — 2026-08-10
+
+Schema: 20.
+
+### Added
+- **File version history** (`pdfs-fuse/src/revisions.rs` + a GTK versions dialog): list a
+  file's revisions, restore one, delete one, or write an old one out to a local path. A
+  restore re-points the file server-side, so no content crosses the wire and nothing enters
+  the drain queue; the daemon drops its cached content and repopulates on the next read.
+- **Persistent SDK entity cache** (`pdfs-core/src/sdkcache.rs`): the SDK's decrypted node
+  metadata now survives a daemon restart instead of re-walking the tree. It lives in its own
+  `sdk_cache.db`, deliberately off the daemon's hot `cache.db` mutex, holds no key material,
+  needs no migrations, and is cleared on logout.
+- Per-node outcomes for batch operations (`pdfs-core/src/batch.rs`): `trash`/`restore`/
+  `delete`/`move` report one result per node, so a node the server rejects no longer fails
+  the whole batch.
+- Photo viewer page in the GUI and richer photo metadata in the details pane.
+
+### Changed
+- SDK telemetry routed into `tracing`, so its transfer/block-storage/request spans land in
+  the daemon log instead of a no-op sink.
+- A file that fits in one block uploads as a single atomic request rather than the
+  draft/block/commit sequence.
+
+## [1.3.0] — 2026-07-31
+
+Schema: 19.
+
+### Added
+- **Photo albums** — `PhotoAlbums` / `AlbumPhotos` control requests, daemon-side `albums.rs`
+  and a GUI albums page (the former gallery view, renamed). Albums other people share with
+  you are included; their photos are persisted per album because a shared album's photos
+  never appear in your own timeline. Served stale-while-refreshing like the timeline.
+- **GUI "Locations" page** — one row per mount spec: the primary `~/ProtonDrive` mount
+  alongside every folder this computer backs up, mirrored or on-demand. Called *Locations*
+  rather than *Mounts* because a mirror folder is a plain directory with no FUSE session.
+  It absorbs the old Computers page's actions and the mountpoint chooser from Settings.
+- `role` on shared-node metadata (`viewer`/`editor`/`admin`), defaulted for wire-compat with
+  older clients and daemons.
+- FUSE acceptance coverage for the permission-based mounts.
+
+## [1.2.1] — 2026-07-31
+
+Schema: 18.
+
+### Fixed
+- Migration bug that could leave the `ProtonDrive` mount unwritable after upgrade.
+- Permission error on on-demand mounts.
+- B78.
+
+### Changed
+- Release pipeline: `.deb` signing added and `.rpm` signing fixed.
+
+## [1.2.0] — 2026-07-29
+
+Sharing and multi-mount release. Schema: 18.
+
+### Added
+- **Shared with me**: shares others sent you are listed, verified and mountable, including on
+  the primary mount.
+  They appear under a synthetic `Shared with me` directory on a virtual volume
+  (`pdfs-fuse/src/virtual.rs`), with name collisions and over-long components resolved by
+  its own naming rules.
+- Per-share access model (`db/share_access.rs`): the effective role of each shared root is
+  persisted and kept resident in mount state, so interning an inode never costs a query. The
+  database stays the restart/offline authority — every role change is written through before
+  the cache updates. A share you cannot write is presented POSIX read-only.
+- Unified location model (`pdfs-core/src/mounts.rs`, `db/mounts.rs`): one presentation shape
+  for every local Proton Drive location, joining device sync state with a new `mount` table.
+  `MountMode` is `Mirror` (full local copy, reconciled both ways) or `OnDemand` (FUSE session
+  fetching on access), with an `Unknown` fallback so a mode written by a newer client does
+  not break an older one.
+
+## [1.1.1] — 2026-07-27
+
+Schema: 16.
+
+### Fixed
+- Trash refresh did not pull updates.
+- Packaging: `PKGBUILD` was left at 1.1.0 by the 1.1.1 version bump, failing the release
+  workflow's tag/manifest/PKGBUILD agreement check.
+
+## [1.1.0] — 2026-07-26
+
+Conflict-handling and concurrency release. Schema: 16.
+
+### Added
+- Conflict sweep (`pdfs-fuse/src/sweep.rs`, 5-minute cadence): removes conflict copies proven
+  identical by size + `content_sha1`, surfaces divergent ones once as an activity entry, and
+  never removes what it cannot prove.
+- Sweep mode and `.pdfsignore`-driven conflict handling; ignore rules are rebuilt per sync
+  pass so edits take effect without a restart.
+- Substantially extended FUSE acceptance suite.
+
+### Fixed
+- **B69** — spurious `(sync-conflict)` copies. Revision identity now uses the server's
+  `revision_id` rather than `(mtime, size)`, which the server re-stamps on seal; the planner
+  gained `AdoptBaseline` so an equal-size no-baseline pair is recorded instead of forked on
+  first reconcile.
+- **B70** — browser downloads into an on-demand mount had their in-flight temporaries sealed
+  as canonical revisions. Transient names are recognised and held until the name is finalized.
+- Assorted race conditions and concurrency bugs on the write/drain paths.
+
+### Changed
+- Workspace deps bumped to `proton-sdk` / `proton-drive-rs` 0.2.2.
+
+## [1.0.0] — 2026-07-22
+
+First stable release: FUSE files-on-demand mount, sync daemon under `proton-drive.service`,
+`pdfs` CLI and the GTK4 app, on top of the pure-Rust Proton Drive SDK. Schema: 16.
+
+### Fixed
+- The outstanding FUSE defects tracked in `docs/BUGS.md`, plus a truncate defect, validated
+  by a new POSIX compliance suite for the filesystem.
+
+[1.7.1]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.7.1
+[1.7.0]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.7.0
+[1.6.0]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.6.0
+[1.5.0]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.5.0
+[1.4.0]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.4.0
+[1.3.0]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.3.0
+[1.2.1]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.2.1
+[1.2.0]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.2.0
+[1.1.1]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.1.1
+[1.1.0]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.1.0
+[1.0.0]: https://github.com/narrrl/proton-drive-linux/releases/tag/v1.0.0

docs/plans/search-index.md

@@ -1,110 +0,0 @@
-# Search index + prompt display fixes
-
-Working plan. Findings come from an audit of a real `cache.db`
-(9,252 Drive nodes, 110,187 local files) on 2026-08-12.
-
-## Evidence
-
-```
-nodes                     9252
-nodes_fts                 6547     ← 2705 rows never indexed
-nodes in orphan subtrees  2705     ← exactly the missing set
-local_files              110187
-  /home/narl/go/pkg/mod   63399    (58% of the local index)
-query "test"             46405 local matches, candidate pool capped at 1000
-  top 500 by fts rank      367 (73%) from go/pkg/mod
-```
-
-`pdfs search tickets` returns only `Documents/Tickets` (My Files); the
-`tickets.pdf` sitting in the device folder `Downloads` is invisible.
-
-## P1 — 29% of Drive is unsearchable
-
-`node_is_indexable_tx` (`crates/pdfs-core/src/db/nodes.rs:606`) indexes a node
-only when the ancestor walk terminates at a row with `parent_uid IS NULL`. The
-DB holds exactly one such row (the My Files root). Device roots live in
-`device`/`sync_folder`, never in `nodes`, so every device-folder subtree
-(`Downloads`, `Pictures`, `Documents`, `Music`, `Videos` under device
-*nthinkpad*) walks up to a `parent_uid` with no row and is judged unindexable.
-Stale `pdfs-acceptance-*` roots hit the same path.
-
-- [x] Relax the rule to "the walk terminates and no ancestor is trashed",
-      keeping the existing cycle guard. A subtree whose top row's parent is
-      simply not cached is still real, reachable data.
-- [x] `MIGRATION_V21`: rebuild `nodes_fts` under the new rule, so existing
-      installs recover the missing rows instead of waiting for each node to be
-      upserted again.
-- [x] Tests: a node under an uncached parent is indexed; a trashed ancestor
-      still excludes; the migration backfills.
-
-### Found while testing: a parent cycle hangs the daemon
-
-Writing the cycle test surfaced a live defect, not a test artefact. A query
-below the trigram minimum takes the `LIKE` lane, which reads `nodes` directly
-and never consults the index, and every row it returns goes through `path_of` —
-whose parent walk is an unguarded `UNION ALL`. On a `parent_uid` cycle it never
-terminates, while holding the daemon's only SQLite connection.
-
-- [x] Depth-cap `path_of` (`db/utils.rs`), matching the 1024 cap
-      `path_relative_to` already carried.
-- [x] Depth-cap the `descendants` CTE in `upsert_node_tx`, which has the same
-      exposure on the write side.
-- [x] Keep cycles out of the index (part of the P1 rule above).
-
-Still unguarded, not touched here: `pin_is_pinned`'s ancestor walk.
-
-## P2 — local index noise crowds real files out of the candidate pool
-
-`SKIP_DIRS` (`crates/pdfs-core/src/localindex.rs:23`) misses the Go module
-cache and build output dirs. `search_v2` clamps candidates to ≤1000 and scores
-*after* SQLite, so for any common term the pool is exhausted by vendored files
-before a real one is considered.
-
-- [x] Skip `go/pkg/mod` (path-relative, not a bare name), `dist`, `build`,
-      `.cache`-style artefact dirs.
-- [x] Give `search_local_candidates` a name-prefix lane, mirroring what
-      `search_candidates` already does for Drive, so an exact-prefix match
-      cannot be starved by trigram rank.
-- [x] Tests for both.
-
-## P3 — display lies about where a hit lives
-
-- [x] `Hit::location()` (`crates/pdfs-gui/src/prompt.rs:150`) hardcodes
-      `My files / …` for every Drive hit. Derive the label from
-      `mounted_path` when the daemon resolved one (device folder, sync folder)
-      and fall back to `My files` only for the primary mount.
-- [x] Pin rows set `is_dir: pin.recursive` (`dmenu.rs:133`, `prompt.rs:872`) —
-      a non-recursively pinned folder renders and activates as a file.
-- [ ] `pdfs locations` prints `Device folder 1..5` rather than the device and
-      folder name. (Cosmetic; deferred.)
-
-## Verification
-
-Done, 2026-08-12:
-
-```
-cargo fmt --all -- --check                                  clean
-cargo clippy --workspace --all-targets -- -D warnings       clean
-cargo test --workspace --locked                             439 passed, 0 failed
-```
-
-The V21 backfill was also replayed against a read-only `.backup` of the live
-`cache.db`:
-
-```
-live_nodes         9276
-indexed            9276     (was 6547)
-downloads_subtree   300
-tickets.pdf ->     Downloads/tickets.pdf
-```
-
-That run is what caught the last defect: the backfill originally seeded an
-uncached-parent root with an empty path, so a node indexed by the migration
-(`tickets.pdf`) and the same node after a later upsert (`Downloads/tickets.pdf`)
-disagreed. The seed now keeps the name for any row that has a parent, matching
-`path_of`.
-
-Still to do, on the live system: rebuild, restart `proton-drive.service`, and
-confirm `pdfs search tickets` returns the device folder's `tickets.pdf`. The
-schema move to 21 is one-way — an older `pdfs` refuses a v21 database. The
-`local_files` shrink (~110k → ~47k) lands with the next background scan.

packaging/PKGBUILD

@@ -9,7 +9,7 @@
 # Build + install:   cd packaging && makepkg -fi
 
 pkgname=proton-drive-linux
-pkgver=1.7.0
+pkgver=1.8.0
 pkgrel=1
 pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray"
 arch=('x86_64')