From 098da2e97a4c023a05cb009013507c267616cf65 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Sun, 16 Aug 2026 21:34:12 +0200 Subject: [PATCH] fix(log) log unreachable nodes --- crates/pdfs-core/src/db/migrations.rs | 41 ++++- crates/pdfs-core/src/db/ops.rs | 61 +++++++- crates/pdfs-core/src/db/tests.rs | 54 +++++++ crates/pdfs-fuse/src/drain.rs | 216 ++++++++++++++++++++++++-- 4 files changed, 353 insertions(+), 19 deletions(-) diff --git a/crates/pdfs-core/src/db/migrations.rs b/crates/pdfs-core/src/db/migrations.rs index 2676227..b6273a5 100644 --- a/crates/pdfs-core/src/db/migrations.rs +++ b/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; +"; diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index 470ba9e..89b5091 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/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 { + 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 diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index 5d566a9..febd459 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/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. diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 1f13885..dab957a 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/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] {