fix(log) log unreachable nodes

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 = 26;
+pub(super) const SCHEMA_VERSION: i64 = 27;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -182,6 +182,25 @@ impl Db {
                 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",
@@ -769,3 +788,23 @@ 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/ops.rs

@@ -721,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],
         )?;
@@ -751,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.

crates/pdfs-fuse/src/drain.rs

@@ -48,6 +48,29 @@ use super::{
 
 const DRAIN_ACCESS_RECHECK: Duration = Duration::from_secs(5);
 
+/// 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 —
+/// notably a node the local tree no longer has a row for, which
+/// [`Db::effective_node_access`] cannot distinguish from a revoked permission
+/// and so reports as `EACCES` forever. 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.
+///
+/// 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 {
@@ -235,25 +258,16 @@ impl Core {
                 || 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.release_claim(&op);
-                    self.wait_for_drain_work();
-                } else {
-                    debug!(
-                        uid = %op.uid,
-                        next_attempt_at,
-                        "pending operation deferred until access is writable"
-                    );
-                    self.release_claim(&op);
-                }
+                self.defer_for_access(&op);
                 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();
@@ -285,6 +299,76 @@ impl Core {
         }
     }
 
+    /// 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) {
+        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,
+                "pending operation deferred: its node is not writable"
+            );
+        }
+        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!(
+                "not writable locally for {}s; the node is missing from the local tree or the \
+                 share is no longer writable. The staged bytes are kept.",
+                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(),
+                "pending operation deferred until access is writable"
+            );
+        }
+        self.release_claim(op);
+    }
+
     /// Hand a claimed op back to the queue, logging rather than propagating —
     /// there is no caller left to propagate to, and the drain must not stop.
     fn release_claim(&self, op: &PendingOp) {
@@ -1494,6 +1578,104 @@ 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 owner_and_editor_drain_authorities_apply_normally() {
         for access in [Access::Owner, Access::Editor] {