diff --git a/Cargo.lock b/Cargo.lock index 4959bda..6965f78 100644 --- a/Cargo.lock +++ b/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.7.1" dependencies = [ "anyhow", "clap 4.6.4", @@ -3533,7 +3553,7 @@ dependencies = [ [[package]] name = "pdfs-core" -version = "1.7.0" +version = "1.7.1" dependencies = [ "async-trait", "directories", @@ -3554,7 +3574,7 @@ dependencies = [ [[package]] name = "pdfs-fuse" -version = "1.7.0" +version = "1.7.1" dependencies = [ "fuser", "futures", @@ -3571,7 +3591,7 @@ dependencies = [ [[package]] name = "pdfs-gui" -version = "1.7.0" +version = "1.7.1" 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", ] diff --git a/Cargo.toml b/Cargo.toml index 6161973..a440730 100644 --- a/Cargo.toml +++ b/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.7.1" edition = "2024" license = "MIT" authors = ["Nils Pukropp "] 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" } diff --git a/crates/pdfs-core/src/db/migrations.rs b/crates/pdfs-core/src/db/migrations.rs index 1baebb7..2676227 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 = 25; +pub(super) const SCHEMA_VERSION: i64 = 26; impl Db { pub(super) fn migrate(&self) -> Result<()> { @@ -161,6 +161,27 @@ impl Db { tx.execute_batch(MIGRATION_V25)?; } } + if current < 26 { + // Same guard as V22: a fixture that never exercised the queue may + // not have `pending_op`, and altering an absent table is an error + // rather than a no-op. + // A fixture built from the current schema and then rewound to an + // older `schema_version` already has the column, and `ADD COLUMN` + // has no `IF NOT EXISTS`. + let has_column: bool = tx.query_row( + "SELECT COUNT(*) FROM pragma_table_info('pending_op') WHERE name = 'claimed_at'", + [], + |row| row.get::<_, i64>(0), + )? > 0; + let has_ops: bool = tx.query_row( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'pending_op'", + [], + |row| row.get::<_, i64>(0), + )? > 0; + if has_ops && !has_column { + tx.execute_batch(MIGRATION_V26)?; + } + } tx.execute( "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1) ON CONFLICT(key) DO UPDATE SET value = excluded.value", @@ -725,3 +746,26 @@ UPDATE nodes DROP TABLE node_paths; "; + +/// V26: `pending_op.claimed_at` — the drain's claim column. +/// +/// The drain was one thread taking one op at a time, so a 10 GiB upload held +/// every queued rename and small write behind it for as long as it ran. Several +/// workers can share the queue only if a claim is visible to the others: this +/// column is what makes "already being drained by someone" a fact about the row +/// rather than about one thread's local variable. +/// +/// `0` is unclaimed; anything else is the ms at which a worker took it. A +/// claim is process-local state that outlives nothing — the single-writer lock +/// means a non-zero value found at open belongs to a crashed run, and +/// [`Db::clear_op_claims`](super::Db::clear_op_claims) drops the lot. +/// +/// The index is partial so it costs nothing on the ordinary row: only the +/// handful of claimed rows are in it, which is exactly the set the claim query +/// has to subtract to keep two workers off one node. +const MIGRATION_V26: &str = " +ALTER TABLE pending_op ADD COLUMN claimed_at INTEGER NOT NULL DEFAULT 0; + +CREATE INDEX IF NOT EXISTS idx_pending_op_claimed + ON pending_op(uid) WHERE claimed_at <> 0; +"; diff --git a/crates/pdfs-core/src/db/mod.rs b/crates/pdfs-core/src/db/mod.rs index 4b3d86d..63e2da3 100644 --- a/crates/pdfs-core/src/db/mod.rs +++ b/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. diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index 041e244..470ba9e 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/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> { + 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 { + 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> { let conn = self.read(); let ts: Option = 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, ), diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index 55caac7..5d566a9 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -3545,3 +3545,114 @@ fn a_read_does_not_wait_for_the_write_connection() { let _ = std::fs::remove_file(&path); let _ = std::fs::remove_file(format!("{}.lock", path.display())); } + +/// §5: the drain is several workers over one queue, and the claim column is the +/// only thing keeping them off each other's work. Two properties matter — no +/// row is ever handed to two workers, and no *node* is ever worked by two +/// workers even though its ops are separate rows. +#[test] +fn a_claimed_op_is_invisible_to_every_other_worker() { + let db = Db::open_in_memory().unwrap(); + let root = uid("root").to_string(); + for i in 0..3 { + db.enqueue_op(&bulk_op(i, &root)).unwrap(); + } + + let first = db.claim_next_due_op(10).unwrap().expect("an op is due"); + let second = db + .claim_next_due_op(10) + .unwrap() + .expect("another op is due"); + let third = db + .claim_next_due_op(10) + .unwrap() + .expect("a third op is due"); + assert!( + db.claim_next_due_op(10).unwrap().is_none(), + "a fourth claim must find nothing left: every queued op is taken" + ); + + let mut ids = [first.id, second.id, third.id]; + ids.sort_unstable(); + assert_eq!( + ids, + [1, 2, 3], + "each worker must get a distinct row, oldest first" + ); + + // Releasing puts one back, and it is the one that comes out next. + db.release_op_claim(second.id).unwrap(); + let again = db.claim_next_due_op(10).unwrap().expect("the released op"); + assert_eq!(again.id, second.id); +} + +#[test] +fn two_workers_never_share_a_node() { + let db = Db::open_in_memory().unwrap(); + let root = uid("root").to_string(); + // Two ops of different kinds against one node — a queued rename and a + // queued revision, which is an ordinary `mv` plus a save. + let mut revision = bulk_op(0, &root); + revision.parent_uid = None; + db.enqueue_op(&revision).unwrap(); + let rename = PendingOp { + kind: OP_RENAME.to_string(), + parent_uid: Some(root.clone()), + blob_path: None, + meta_json: None, + ..bulk_op(0, &root) + }; + db.enqueue_op(&rename).unwrap(); + + let claimed = db.claim_next_due_op(10).unwrap().expect("an op is due"); + assert_eq!(claimed.kind, OP_REVISION, "the older op goes first"); + assert!( + db.claim_next_due_op(10).unwrap().is_none(), + "the second op targets the same node, so it must wait for the first: \ + two workers on one uid would race over its staged blob and land its \ + changes out of order" + ); + + db.release_op_claim(claimed.id).unwrap(); + db.delete_op(claimed.id).unwrap(); + assert_eq!( + db.claim_next_due_op(10).unwrap().map(|o| o.kind), + Some(OP_RENAME.to_string()), + "once the revision retires, the rename is claimable" + ); +} + +#[test] +fn claims_do_not_survive_the_process_that_took_them() { + let db = Db::open_in_memory().unwrap(); + let root = uid("root").to_string(); + for i in 0..2 { + db.enqueue_op(&bulk_op(i, &root)).unwrap(); + } + db.claim_next_due_op(10).unwrap().unwrap(); + db.claim_next_due_op(10).unwrap().unwrap(); + assert!(db.claim_next_due_op(10).unwrap().is_none()); + + // What `Db::open` does after the single-writer lock says the previous run + // is gone. Without it those ops are invisible to every worker forever. + assert_eq!(db.clear_op_claims().unwrap(), 2); + assert!( + db.claim_next_due_op(10).unwrap().is_some(), + "a crashed run's claims must not park its queue" + ); +} + +#[test] +fn an_idle_worker_does_not_wait_on_an_op_another_worker_has() { + let db = Db::open_in_memory().unwrap(); + let root = uid("root").to_string(); + db.enqueue_op(&bulk_op(0, &root)).unwrap(); + + assert_eq!(db.earliest_due_at().unwrap(), Some(0)); + db.claim_next_due_op(10).unwrap().unwrap(); + assert_eq!( + db.earliest_due_at().unwrap(), + None, + "an op somebody else is draining is not work this worker is waiting for" + ); +} diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 26444cd..1f13885 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -20,10 +20,14 @@ //! are kept as a conflict copy rather than discarded — see //! [`Core::revision_conflict`]. +use std::collections::HashMap; use std::fs::File; use std::path::{Path, PathBuf}; -use std::sync::atomic::Ordering; -use std::time::Duration; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant}; + +use parking_lot::Mutex; use pdfs_core::batch; use pdfs_core::cache::{Baseline, StagedWrite}; @@ -36,13 +40,59 @@ use tracing::{debug, error, info, warn}; use super::state::Intervals; use super::transfers::CountingReader; use super::{ - Core, DRAIN_BACKOFF_MAX, DRAIN_BACKOFF_MIN, DRAIN_IDLE_POLL, ROOT_INO, conflict_name, - is_already_exists, is_gone, is_local_uid_str, media_type_for, node_revision_id, node_size, - now_millis, now_secs, parse_node_uid, + Core, DRAIN_BACKOFF_MAX, DRAIN_BACKOFF_MIN, DRAIN_IDLE_POLL, DRAIN_REVISION_DEBOUNCE, + DRAIN_REVISION_DEBOUNCE_MAX, ROOT_INO, UPLOAD_TIME_MEMORY, conflict_name, is_already_exists, + is_gone, is_local_uid_str, media_type_for, node_revision_id, node_size, now_millis, now_secs, + parse_node_uid, }; const DRAIN_ACCESS_RECHECK: Duration = Duration::from_secs(5); +/// A registered cancellation flag for one in-flight upload, deregistered when +/// dropped. See [`Core::begin_cancellable_upload`]. +struct UploadCancel { + registry: Arc>>>, + uid: NodeUid, + flag: Arc, +} + +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 { + 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 { + measured + .unwrap_or(DRAIN_REVISION_DEBOUNCE) + .clamp(DRAIN_REVISION_DEBOUNCE, DRAIN_REVISION_DEBOUNCE_MAX) +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum DrainDisposition { Applied, @@ -127,24 +177,57 @@ impl Core { /// that is due — one file wedged against a folder that no longer exists must /// not hold up an unrelated upload behind it. The worker only blocks once /// nothing is due at all. - pub(crate) fn run_pending_drain(&self) { + /// + /// Several of these run at once ([`DRAIN_WORKERS`]), sharing the queue + /// through [`Db::claim_next_due_op`]: the alternative was that a 10 GiB + /// upload held every queued rename, trash and small write behind it for as + /// long as it took. Ordering only has to hold *per node*, and the claim + /// query is what guarantees it — no two workers are ever on one uid. + /// + /// `primary` marks the one worker that also runs the queue's idle chores. + /// They are cheap, but they are not per-op work and doing them in every + /// worker would just multiply them. + /// + /// [`DRAIN_WORKERS`]: super::DRAIN_WORKERS + /// [`Db::claim_next_due_op`]: pdfs_core::db::Db::claim_next_due_op + pub(crate) fn run_pending_drain(&self, primary: bool) { loop { let now = now_millis(); - // One row, chosen by the database. Reading the whole queue to pick - // one op made a long queue quadratic to drain, and held the shared - // connection — and so every FUSE metadata call — for the duration. - let due = self.db.next_due_op(now).unwrap_or_default(); - - let Some(op) = due.filter(|_| self.online.load(Ordering::Relaxed)) else { - // Nothing to upload is exactly when the connection is free, so - // this is where the read path's buffered LRU touches get written - // (`ContentCache::flush_touches`). Cheap and a no-op when empty. - self.cache.flush_touches(); - self.recover_fsynced_writes(); - // A debounced or backed-off op may be waiting: sleep only - // until it becomes due rather than the full idle-poll. - self.wait_for_drain_work_or_due(); - continue; + // One row, chosen and *claimed* by the database. Reading the whole + // queue to pick one op made a long queue quadratic to drain, and + // held the shared connection — and so every FUSE metadata call — + // for the duration. + let due = match self.db.claim_next_due_op(now) { + Ok(due) => due, + Err(error) => { + error!(%error, "claiming the next pending operation failed"); + self.wait_for_drain_work(); + continue; + } + }; + let online = self.online.load(Ordering::Relaxed); + let op = match due { + Some(op) if online => op, + idle => { + // Claimed but offline: hand it straight back, or it stays + // invisible to every worker until this one happens to be + // the next to notice the reconnect. + if let Some(op) = idle { + self.release_claim(&op); + } + if primary { + // Nothing to upload is exactly when the connection is + // free, so this is where the read path's buffered LRU + // touches get written (`ContentCache::flush_touches`). + // Cheap and a no-op when empty. + self.cache.flush_touches(); + self.recover_fsynced_writes(); + } + // A debounced or backed-off op may be waiting: sleep only + // until it becomes due rather than the full idle-poll. + self.wait_for_drain_work_or_due(); + continue; + } }; let outcome = run_authorized_drain( &op, @@ -159,6 +242,7 @@ impl Core { %error, "deferring access-blocked pending operation failed" ); + self.release_claim(&op); self.wait_for_drain_work(); } else { debug!( @@ -166,6 +250,7 @@ impl Core { next_attempt_at, "pending operation deferred until access is writable" ); + self.release_claim(&op); } continue; } @@ -188,9 +273,103 @@ impl Core { now_millis() + backoff.as_millis() as i64, ) { error!(uid = %op.uid, error = %e, "recording a drain failure failed"); + self.release_claim(&op); self.wait_for_drain_work(); + continue; } } + // Unconditionally, on every path: a handler that retired its own row + // leaves nothing to release, and one that returned without retiring + // must not leave the row claimed by a worker that has moved on. + self.release_claim(&op); + } + } + + /// Hand a claimed op back to the queue, logging rather than propagating — + /// there is no caller left to propagate to, and the drain must not stop. + fn release_claim(&self, op: &PendingOp) { + if let Err(error) = self.db.release_op_claim(op.id) { + error!(uid = %op.uid, %error, "releasing a drain claim failed"); + } + } + + /// How long a freshly queued revision of `uid` should wait before a worker + /// may pick it up. + /// + /// [`DRAIN_REVISION_DEBOUNCE`] exists for a tool that preallocates a file + /// and then writes it, where two seconds is the right scale. It is the wrong + /// scale for a large file on a slow link: an editor saving a 200 MB project + /// every ten seconds queues a write, the drain starts sending it, the next + /// save supersedes it mid-flight, and the file never finishes uploading + /// while the user keeps working. Waiting roughly as long as the last upload + /// of *this* node took moves that supersede into the queue, where replacing + /// a row costs nothing. + /// + /// Never shorter than the fixed debounce and never longer than + /// [`DRAIN_REVISION_DEBOUNCE_MAX`]: a node whose upload is genuinely slow + /// still has to reach Drive, and a single pathological stall must not park + /// it. + /// + /// [`DRAIN_REVISION_DEBOUNCE`]: super::DRAIN_REVISION_DEBOUNCE + /// [`DRAIN_REVISION_DEBOUNCE_MAX`]: super::DRAIN_REVISION_DEBOUNCE_MAX + pub(crate) fn revision_debounce(&self, uid: &NodeUid) -> Duration { + let measured = self + .upload_times + .lock() + .get(&uid.to_string()) + .map(|&(ms, _)| Duration::from_millis(ms)); + adaptive_debounce(measured) + } + + /// Remember how long an upload of `uid` took, for the next write's + /// debounce. + /// + /// Only a *completed* upload is a measurement. A cancelled or failed one + /// says how long it ran before it stopped, which is not how long the file + /// takes to send, and feeding that back would let one network fault widen + /// the debounce for good. + fn record_upload_time(&self, uid: &NodeUid, took: Duration) { + let mut times = self.upload_times.lock(); + let now = now_millis(); + if times.len() >= UPLOAD_TIME_MEMORY + && !times.contains_key(&uid.to_string()) + && let Some(oldest) = times + .iter() + .min_by_key(|(_, (_, at))| *at) + .map(|(k, _)| k.clone()) + { + times.remove(&oldest); + } + times.insert(uid.to_string(), (took.as_millis() as u64, now)); + } + + /// Register a cancellation flag for an upload of `uid` that is about to + /// start, returning a guard that deregisters it. + /// + /// Deregistration is by identity rather than by key: a supersede that + /// arrives just as this upload finishes may already have registered the + /// *next* upload's flag under the same uid, and removing that one would + /// leave the newer upload uncancellable. + fn begin_cancellable_upload(&self, uid: &NodeUid) -> UploadCancel { + let flag = Arc::new(AtomicBool::new(false)); + self.upload_cancel.lock().insert(uid.clone(), flag.clone()); + UploadCancel { + registry: self.upload_cancel.clone(), + uid: uid.clone(), + flag, + } + } + + /// Tell an upload of `uid` that is on the wire to stop, if there is one. + /// + /// Called when a newer revision of the file is queued. The upload does not + /// stop instantly — it stops when the SDK next asks its reader for bytes, + /// which for a 4 MiB block is soon enough to matter and late enough to be + /// safe: the staged blob it is reading stays open until it lets go. + pub(crate) fn cancel_upload(&self, uid: &NodeUid) { + if let Some(flag) = self.upload_cancel.lock().get(uid) { + flag.store(true, Ordering::Relaxed); + debug!(%uid, "cancelling a superseded upload"); } } @@ -907,15 +1086,37 @@ impl Core { let guard = self.transfers .begin(&name, meta.uid.clone(), TransferDirection::Upload, meta.len); - let reader = CountingReader::new(File::open(&blob)?, &guard); - self.rt.block_on(self.client.upload_new_revision_from( + // Registered before the first byte moves, so a write that lands while + // this is on the wire can stop it rather than paying for a revision the + // op it queued will immediately replace. + let cancel = self.begin_cancellable_upload(&uid); + let reader = CountingReader::new(File::open(&blob)?, &guard).with_cancel(cancel.flag()); + let started = Instant::now(); + let sent = self.rt.block_on(self.client.upload_new_revision_from( &uid, reader, meta.len as i64, Vec::new(), None, - ))?; + )); + let took = started.elapsed(); + if let Err(e) = sent { + if cancel.cancelled() { + // Not a failure, and not this op's row to retire: the write that + // cancelled us deleted it and took ownership of the staged blob + // when it superseded this revision. Leaving the blob alone here + // is the whole point — it is either already unlinked or it + // belongs to the newer op. + info!(%uid, "upload abandoned; a newer write superseded it"); + return Ok(()); + } + return Err(e.into()); + } + drop(cancel); drop(guard); + // Only a completed upload times how long this file takes to send, which + // is what the next write's debounce is scaled from. + self.record_upload_time(&uid, took); // Retire the op before dropping the blob: a crash between the two leaves // an orphaned file (harmless), whereas the reverse would leave a queued @@ -1413,3 +1614,42 @@ mod tests { assert!(!is_own_self_supersede(true, None, None)); } } + +#[cfg(test)] +mod debounce_tests { + use super::*; + + #[test] + fn a_node_never_uploaded_keeps_the_fixed_grace_period() { + assert_eq!(adaptive_debounce(None), DRAIN_REVISION_DEBOUNCE); + } + + #[test] + fn a_fast_upload_does_not_shorten_the_grace_period() { + // The 2 seconds are not about upload time at all — they are the window + // in which a preallocate-then-write tool supersedes its own first + // close. A file that sends in 50 ms must still get them. + assert_eq!( + adaptive_debounce(Some(Duration::from_millis(50))), + DRAIN_REVISION_DEBOUNCE + ); + } + + #[test] + fn a_slow_upload_widens_the_grace_period_to_match() { + // The case the fixed debounce got wrong: saves arriving faster than the + // upload completes, each superseding one already on the wire. + assert_eq!( + adaptive_debounce(Some(Duration::from_secs(25))), + Duration::from_secs(25) + ); + } + + #[test] + fn a_pathological_upload_cannot_park_a_node() { + assert_eq!( + adaptive_debounce(Some(Duration::from_secs(6000))), + DRAIN_REVISION_DEBOUNCE_MAX + ); + } +} diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index b4afe94..ea52976 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/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>>, - /// 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, 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>>, + /// 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>>>, /// True while a background refresh of the photos timeline (resp. the trash) is /// already running, so a burst of page requests against a stale listing kicks /// off one refresh rather than one per request. @@ -2756,6 +2794,12 @@ impl Core { } })?; let meta_json = serde_json::to_string(&meta).unwrap_or_default(); + // Whatever this write supersedes, it supersedes now — including an + // upload already on the wire, whose bytes describe a revision the row + // written just below replaces. Signalled before the enqueue so the + // in-flight reader stops at its next block rather than after however + // much of the file is left. + self.cancel_upload(uid); let superseded = if is_local_uid(uid) { // The node has no server-side identity to hang a revision on, so the // bytes ride on the create that will mint it. @@ -2789,7 +2833,7 @@ impl Core { created_at: now_millis(), attempts: 0, last_error: None, - next_attempt_at: now_millis() + DRAIN_REVISION_DEBOUNCE.as_millis() as i64, + next_attempt_at: now_millis() + self.revision_debounce(uid).as_millis() as i64, }; let (_id, superseded) = self.db.enqueue_op(&op).map_err(|e| { error!(%uid, error = %e, "queueing upload failed"); @@ -3095,6 +3139,12 @@ impl Core { /// worse in every way. fn queue_trash(&self, uid: &NodeUid, name: &str) -> Result<(), Errno> { self.require_uid_writable(uid)?; + // The bytes below are about to be discarded, and a drain worker may be + // reading them onto the wire right now. Stop it first: with several + // workers the trash op this queues can be claimed while that upload is + // still running, and the two would race to decide whether the file + // exists. + self.cancel_upload(uid); let (_, blobs) = self .db .replace_ops_with_trash(&uid.to_string(), name, now_millis()) @@ -3124,6 +3174,9 @@ impl Core { /// Drop every op queued against a node, and the staged bytes they own. fn discard_queued_ops(&self, uid: &NodeUid) -> Result<(), Errno> { + // Same reason as `queue_trash`: an upload may be reading the very blobs + // this is about to unlink. + self.cancel_upload(uid); let blobs = self.db.delete_ops_for_uid(&uid.to_string()).map_err(|e| { error!(%uid, error = %e, "dropping queued ops failed"); Errno::EIO @@ -6876,6 +6929,19 @@ mod tests { let trash = function_source(source, "fn queue_trash("); assert_before(trash, "replace_ops_with_trash", "discard_staged"); + // §5, parallel drain: every path that unlinks a staged blob or hands + // the same node's work to another worker has to stop the upload that + // may be reading those bytes *first*. Late is a superseded revision on + // the wire, or two workers deciding whether the file exists. + for (signature, mutation) in [ + ("fn enqueue_staged_write(", "attach_blob_to_create"), + ("fn queue_trash(", "replace_ops_with_trash"), + ("fn discard_queued_ops(", "delete_ops_for_uid"), + ] { + let function = function_source(source, signature); + assert_before(function, "cancel_upload", mutation); + } + let filesystem = include_str!("filesystem.rs"); for signature in ["fn serve_unlink(", "fn serve_rmdir("] { let function = function_source(filesystem, signature); diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 9b942c5..fae5699 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/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 diff --git a/crates/pdfs-fuse/src/transfers.rs b/crates/pdfs-fuse/src/transfers.rs index 60264e8..ae8093e 100644 --- a/crates/pdfs-fuse/src/transfers.rs +++ b/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 Read for OwnedCountingReader { } } -/// 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>, } 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) -> Self { + self.cancel = Some(cancel); + self } } impl Read for CountingReader<'_, R> { fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + 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) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index e86a756..7e6228a 100644 --- a/docs/ARCHITECTURE.md +++ b/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 diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index ee71277..2a10f23 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -9,6 +9,26 @@ release ships. Migrations are forward-only — a database written by a newer cli 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.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 @@ -232,6 +252,7 @@ First stable release: FUSE files-on-demand mount, sync daemon under `proton-driv - 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 diff --git a/packaging/PKGBUILD b/packaging/PKGBUILD index 272d742..64e40fd 100644 --- a/packaging/PKGBUILD +++ b/packaging/PKGBUILD @@ -9,7 +9,7 @@ # Build + install: cd packaging && makepkg -fi pkgname=proton-drive-linux -pkgver=1.7.0 +pkgver=1.7.1 pkgrel=1 pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray" arch=('x86_64')