diff --git a/crates/pdfs-core/src/db/nodes.rs b/crates/pdfs-core/src/db/nodes.rs index 4c2842b..9906d3a 100644 --- a/crates/pdfs-core/src/db/nodes.rs +++ b/crates/pdfs-core/src/db/nodes.rs @@ -559,6 +559,15 @@ impl Db { if listed != Some(1) { return Ok(None); } + drop(conn); + self.known_children(parent).map(Some) + } + + /// The children the DB knows under `parent`, whether or not the folder's + /// listing is still marked complete. Only for when nothing better can be + /// had: a listing dropped as stale may miss or keep a child. + pub fn known_children(&self, parent: &NodeUid) -> Result> { + let conn = self.read(); let mut stmt = conn.prepare( "SELECT node_json FROM nodes WHERE parent_uid = ?1 AND node_json IS NOT NULL AND trashed = 0", @@ -568,7 +577,7 @@ impl Db { for json in rows { out.push(serde_json::from_str(&json?)?); } - Ok(Some(out)) + Ok(out) } // --- Content-cache LRU index (P4) ------------------------------------- diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index 91a74e3..592bb96 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/crates/pdfs-core/src/db/ops.rs @@ -220,6 +220,24 @@ pub struct AttachedBlob { } impl Db { + /// Whether a queued trash still holds `name` somewhere. + /// + /// A replacing rename or a delete-then-recreate queues the trash of the old + /// node ahead of the op that wants its name. With several drain workers the + /// later op can run first and find the name still taken; this tells it to + /// wait for the trash rather than land under a conflict name. + pub fn has_pending_trash_named(&self, name: &str) -> Result { + let conn = self.read(); + conn.query_row( + "SELECT EXISTS( + SELECT 1 FROM pending_op WHERE kind = ?1 AND name = ?2 + )", + params![OP_TRASH, name], + |row| row.get(0), + ) + .map_err(Into::into) + } + /// Whether a specific desired-state operation is still queued for a node. pub fn has_pending_op(&self, uid: &str, kind: &str) -> Result { let conn = self.read(); diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index ab46761..996cfef 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -788,6 +788,8 @@ fn children_if_listed_gated_on_flag() { // Not listed yet → unknown, force a re-fetch. assert!(db.children_if_listed(&uid("root")).unwrap().is_none()); + // Still known, for a caller that has nothing better while offline. + assert_eq!(db.known_children(&uid("root")).unwrap().len(), 2); db.set_listed(&uid("root"), true).unwrap(); let kids = db.children_if_listed(&uid("root")).unwrap().unwrap(); @@ -2458,6 +2460,20 @@ fn atomic_trash_replacement_returns_owned_blobs_after_commit() { assert!(ops[0].blob_path.is_none()); } +#[test] +fn a_queued_trash_holds_its_name_until_it_lands() { + let db = Db::open_in_memory().unwrap(); + let node = uid("held-name").to_string(); + assert!(!db.has_pending_trash_named("report.txt").unwrap()); + + let (trash_id, _) = db.replace_ops_with_trash(&node, "report.txt", 1).unwrap(); + assert!(db.has_pending_trash_named("report.txt").unwrap()); + assert!(!db.has_pending_trash_named("other.txt").unwrap()); + + db.delete_op(trash_id).unwrap(); + assert!(!db.has_pending_trash_named("report.txt").unwrap()); +} + #[test] fn failed_atomic_trash_insert_keeps_prior_revision_and_blob_ownership() { let db = Db::open_in_memory().unwrap(); diff --git a/crates/pdfs-fuse/src/background.rs b/crates/pdfs-fuse/src/background.rs index 3d31953..048c520 100644 --- a/crates/pdfs-fuse/src/background.rs +++ b/crates/pdfs-fuse/src/background.rs @@ -378,9 +378,10 @@ pub(super) async fn run_event_sync( info!(?cursor, "event sync started"); loop { - tokio::time::sleep(POLL_INTERVAL).await; - let events = match client.enumerate_events(&scope, cursor.as_ref()).await { + core.next_poll(POLL_INTERVAL).await; + let events = match link::bounded(client.enumerate_events(&scope, cursor.as_ref())).await { Ok(events) => events, + Err(e) if core.lost_link(&e, "poll events") => continue, Err(e) => { warn!(error = %e, "event poll failed; retrying after interval"); continue; @@ -471,9 +472,10 @@ pub(super) async fn run_photos_event_sync(core: Core) { info!(?cursor, "photo event sync started"); loop { - tokio::time::sleep(POLL_INTERVAL).await; - let events = match photos.enumerate_events(&scope, cursor.as_ref()).await { + core.next_poll(POLL_INTERVAL).await; + let events = match link::bounded(photos.enumerate_events(&scope, cursor.as_ref())).await { Ok(events) => events, + Err(error) if core.lost_link(&error, "poll photo events") => continue, Err(error) => { warn!(%error, "photo event poll failed; retrying after interval"); continue; diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 757e2a4..41bffe8 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -38,7 +38,7 @@ use pdfs_core::db::{ }; use proton_drive_rs::proton_sdk::error::ProtonError; use proton_drive_rs::proton_sdk::ids::NodeUid; -use proton_drive_rs::{Node, NodeMoveItem}; +use proton_drive_rs::{Node, NodeKind, NodeMoveItem}; use tracing::{debug, error, info, warn}; use super::state::{Intervals, PendingRevision}; @@ -673,6 +673,9 @@ impl Core { // *different* name is the non-destructive resolution: it neither // clobbers their file nor drops ours, and it is visible. Err(e) if is_already_exists(&e) => { + if self.db.has_pending_trash_named(&name)? { + return Err(held_by_queued_trash(&name)); + } let alt = conflict_name(&name, now_secs()); warn!(%uid, name, alt, "destination already holds that name; using a conflict name"); self.move_renaming(&uid, &parent, Some(&alt))?; @@ -696,8 +699,8 @@ impl Core { let landed = match self.drain_rename_in_place(&uid, &node.name, &name) { Ok(landed) => landed, // The node went with it; nothing is left to rename. - Err(e) if is_gone(&e) => node.name.clone(), - Err(e) => return Err(e.into()), + Err(e) if is_gone(e.as_ref()) => node.name.clone(), + Err(e) => return Err(e), }; self.log_activity( ActivityKind::Rename, @@ -725,13 +728,16 @@ impl Core { uid: &NodeUid, current: &str, name: &str, - ) -> Result { + ) -> Result> { if current == name { return Ok(current.to_string()); } match self.rename_remote(uid, name) { Ok(()) => Ok(name.to_string()), Err(e) if is_already_exists(&e) => { + if self.db.has_pending_trash_named(name)? { + return Err(held_by_queued_trash(name)); + } let alt = conflict_name(name, now_secs()); warn!(%uid, name, alt, "rename target name is taken; using a conflict name"); self.rename_remote(uid, &alt)?; @@ -744,7 +750,7 @@ impl Core { ); Ok(alt) } - Err(e) => Err(e), + Err(e) => Err(e.into()), } } @@ -842,8 +848,26 @@ impl Core { // reachable only because the op waited, which is the whole point of the // queue. Never overwrite theirs and never drop ours: land under a // conflict name, exactly as the sync engine does. + // + // Two things can hold the name that are not someone else's file. The + // node a create made before the network dropped its answer is ours: the + // create was queued because nobody heard back, so adopt it. And a node + // whose trash is still queued is about to let go of it. let mut name = wanted.clone(); + let mut adopted = false; let mut real = self.create_drained_node(op, &parent, &name); + if real.as_ref().is_err_and(|e| is_already_exists(e.as_ref())) + && let Some(twin) = self.adoptable_twin(op, &parent, &wanted)? + { + info!(%local, %twin, wanted, "the name is held by our own unanswered create; adopting it"); + real = Ok(twin); + adopted = true; + } + if real.as_ref().is_err_and(|e| is_already_exists(e.as_ref())) + && self.db.has_pending_trash_named(&wanted)? + { + return Err(held_by_queued_trash(&wanted)); + } if real.as_ref().is_err_and(|e| is_already_exists(e.as_ref())) { name = conflict_name(&wanted, now_secs()); warn!(%local, wanted, name, "name is taken remotely; creating under a conflict name"); @@ -886,14 +910,17 @@ impl Core { // Retire the op before touching anything else: if we crash here the node // exists remotely and the local placeholder is reconciled by the event // sync, whereas a surviving op would create the file a second time. - let newer = self.retire_create(op, &local, &real)?; + // An adopted node was made empty, so a blob this op carries has not been + // uploaded: it goes on as a revision of the node instead. + let uploaded = (!adopted).then_some(op.blob_path.as_deref()).flatten(); + let newer = self.retire_create(op, uploaded, &local, &real)?; self.adopt_real_uid(&local, &real)?; // The feed will report this create back to us; the tree already has it // under its real uid, so that event is ours to ignore (`Core::self_changes`). self.note_self_change(&real); // The uploaded blob is done with either way: a write that replaced it // while it was on the wire already discarded it when it attached. - if let Some(blob) = op.blob_path.as_deref() { + if let Some(blob) = uploaded { self.cache.discard_staged(Path::new(blob)); } if newer { @@ -905,6 +932,40 @@ impl Core { Ok(()) } + /// The node already holding `name` under `parent`, if it is the one an + /// earlier attempt at this op made before its answer was lost. + /// + /// A create that timed out may still have landed; the op was queued because + /// the caller never heard back. Without this, its replay forked a + /// `(sync-conflict)` copy of a file nobody else had touched. + fn adoptable_twin( + &self, + op: &PendingOp, + parent: &NodeUid, + name: &str, + ) -> Result, Box> { + let uids = self + .rt + .block_on(self.client.enumerate_folder_children_node_uids(parent))?; + let children = self.rt.block_on(self.client.enumerate_nodes_light(&uids))?; + let Some(twin) = children + .into_iter() + .find(|node| node.name == name && !node.trashed) + else { + return Ok(None); + }; + // The light listing carries no file size, and an adoptable file is + // exactly an empty one. + let twin = match twin.kind { + NodeKind::Folder => twin, + NodeKind::File { .. } => match self.fetch_node_remote(&twin.uid)? { + Some(node) => node, + None => return Ok(None), + }, + }; + Ok(adoptable(op.kind == OP_MKDIR, &twin, op.created_at).then_some(twin.uid)) + } + /// Make one queued `create`/`mkdir` real under a given name, and hand back /// the API's own error so the caller can tell a name clash from a failure. /// @@ -981,23 +1042,20 @@ impl Core { fn retire_create( &self, op: &PendingOp, + uploaded: Option<&str>, local: &NodeUid, real: &NodeUid, ) -> Result> { let real_str = real.to_string(); let newer = { let mut pending = self.pending.lock(); - let newer = self.db.finish_create( - op.id, - op.blob_path.as_deref(), - &local.to_string(), - &real_str, - |json| { - let mut meta: StagedWrite = serde_json::from_str(json).ok()?; - meta.uid = real_str.clone(); - serde_json::to_string(&meta).ok() - }, - )?; + let newer = + self.db + .finish_create(op.id, uploaded, &local.to_string(), &real_str, |json| { + let mut meta: StagedWrite = serde_json::from_str(json).ok()?; + meta.uid = real_str.clone(); + serde_json::to_string(&meta).ok() + })?; pending.remove(local); if let Some((blob, json)) = &newer { let meta: StagedWrite = serde_json::from_str(json)?; @@ -1796,6 +1854,31 @@ fn park_verdict(created_at: i64, now: i64, open: bool) -> ParkVerdict { ParkVerdict::Release } +/// The retryable error for an op whose name a queued trash has yet to free. +fn held_by_queued_trash(name: &str) -> Box { + format!("{name} is still held by a node whose trash is queued").into() +} + +/// How much older than its op a remote node may be and still be the one an +/// unanswered call made. The op is queued the moment the call gives up, so the +/// node is at most one call's deadline older; the rest is clock skew between +/// this machine and Proton. +const ADOPT_WINDOW_MS: i64 = 2 * 60 * 1000; + +/// Whether `twin`, found holding the name a queued create wants, is the node an +/// earlier, unanswered attempt at that create made: the same kind, created +/// around when the op was queued, and for a file still empty, as the mount's +/// create leaves it. +fn adoptable(is_dir: bool, twin: &Node, op_created_ms: i64) -> bool { + if twin.creation_time.saturating_mul(1000) < op_created_ms - ADOPT_WINDOW_MS { + return false; + } + match &twin.kind { + NodeKind::Folder => is_dir, + NodeKind::File { claimed_size, .. } => !is_dir && *claimed_size == Some(0), + } +} + #[cfg(test)] mod tests { use super::*; @@ -2175,6 +2258,29 @@ mod tests { } } + #[test] + fn only_a_fresh_empty_twin_of_the_same_kind_is_adopted() { + // The op was queued at t=1000 s, after a create that got no answer. + let queued = 1_000_000; + let made = |mut node: Node, secs: i64| { + node.creation_time = secs; + node + }; + assert!(adoptable(false, &made(file_node(0, 0, None), 990), queued)); + assert!(adoptable(true, &made(folder_node("d", None), 990), queued)); + // Made long before the op: someone else's file, not our lost create. + assert!(!adoptable(false, &made(file_node(0, 0, None), 800), queued)); + // Our create leaves a file empty; one with content is someone else's. + assert!(!adoptable(false, &made(file_node(0, 5, None), 990), queued)); + // The kind has to match what the op makes. + assert!(!adoptable(true, &made(file_node(0, 0, None), 990), queued)); + assert!(!adoptable( + false, + &made(folder_node("d", None), 990), + queued + )); + } + fn baseline(mtime: i64, size: u64, rev: Option<&str>) -> Baseline { Baseline { mtime, diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index 07013d7..cc37d7f 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -1501,47 +1501,55 @@ impl ProtonFs { // // A parent that is itself still queued forces the same path even when we // are online: the API has no folder to put this in yet. - let node = - if !transient && self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid) - { - if let Err(error) = self.core.require_uid_writable(&parent_uid) { - reply.error(error); + // + // A create that set out online and lost the network on the way lands in + // the queue too, so a Wi-Fi drop costs the caller a wait, never an EIO. + let minted = if !transient && self.core.is_online() && !is_local_uid(&parent_uid) { + if let Err(error) = self.core.require_uid_writable(&parent_uid) { + reply.error(error); + return; + } + // Create an empty file on the remote so it has a real uid immediately; + // written bytes are buffered and sealed as a new revision on close. + match self.core.block_on_bounded(self.core.client.upload_file( + &parent_uid, + name, + media_type_for(name), + b"", + )) { + Ok(new_uid) => match self.core.fetch_node(&new_uid) { + Ok(n) => Some(n), + // The file exists remotely but we could not read it back. + // The queued create finds it by name and adopts it. + Err(e) if e == Errno::EIO && !self.core.is_online() => None, + Err(e) => { + reply.error(e); + return; + } + }, + Err(e) if self.core.lost_link(&e, "create") => None, + Err(e) => { + error!(%parent_uid, name, error = %e, "create file failed"); + reply.error(Errno::EIO); return; } - // Create an empty file on the remote so it has a real uid immediately; - // written bytes are buffered and sealed as a new revision on close. - let new_uid = match self.core.rt.block_on(self.core.client.upload_file( - &parent_uid, - name, - media_type_for(name), - b"", - )) { - Ok(u) => u, - Err(e) => { - error!(%parent_uid, name, error = %e, "create file failed"); - reply.error(Errno::EIO); - return; - } - }; - match self.core.fetch_node(&new_uid) { - Ok(n) => n, - Err(e) => { - reply.error(e); - return; - } + } + } else { + None + }; + let node = match minted { + Some(node) => node, + None => match self + .core + .queue_local_node(&parent_uid, name, false, transient) + { + Ok(n) => n, + Err(e) => { + reply.error(e); + return; } - } else { - match self - .core - .queue_local_node(&parent_uid, name, false, transient) - { - Ok(n) => n, - Err(e) => { - reply.error(e); - return; - } - } - }; + }, + }; let new_uid = node.uid.clone(); // The base this handle writes over is the node as it actually exists — // the empty file the server just minted — so its modification time comes @@ -1631,39 +1639,43 @@ impl ProtonFs { // As in `create`: offline — or under a parent that is itself still // queued — the folder becomes a placeholder that the drain turns into a // real one (offline.md Phase 3b). - let node = if self.core.online.load(Ordering::Relaxed) && !is_local_uid(&parent_uid) { + // Losing the network on the way queues it the same way. + let minted = if self.core.is_online() && !is_local_uid(&parent_uid) { if let Err(error) = self.core.require_uid_writable(&parent_uid) { reply.error(error); return; } - let new_uid = - match self - .core - .rt - .block_on(self.core.client.create_folder(&parent_uid, name, now)) - { - Ok(u) => u, + match self + .core + .block_on_bounded(self.core.client.create_folder(&parent_uid, name, now)) + { + Ok(new_uid) => match self.core.fetch_node(&new_uid) { + Ok(n) => Some(n), + Err(e) if e == Errno::EIO && !self.core.is_online() => None, Err(e) => { - error!(%parent_uid, name, error = %e, "create folder failed"); - reply.error(Errno::EIO); + reply.error(e); return; } - }; - match self.core.fetch_node(&new_uid) { - Ok(n) => n, + }, + Err(e) if self.core.lost_link(&e, "mkdir") => None, Err(e) => { - reply.error(e); + error!(%parent_uid, name, error = %e, "create folder failed"); + reply.error(Errno::EIO); return; } } } else { - match self.core.queue_local_node(&parent_uid, name, true, false) { + None + }; + let node = match minted { + Some(node) => node, + None => match self.core.queue_local_node(&parent_uid, name, true, false) { Ok(n) => n, Err(e) => { reply.error(e); return; } - } + }, }; let mut st = self.core.state(); let local = is_local_uid(&node.uid); @@ -1985,12 +1997,11 @@ impl ProtonFs { // take the synchronous path, so a genuine API refusal (permissions, a // name clash) surfaces to the caller instead of becoming a queued op // that can only ever conflict. - if rename_needs_queue( - self.core.online.load(Ordering::Relaxed), - is_local_uid(&new_parent_uid), - newparent != parent, - newname != name, - ) { + // The queued rename, as a closure: the network going away under the + // synchronous path below lands here too, with whatever half already + // reached Drive. The op records the end state, so the drain finishes + // the rest. + let queue_it = |reply: ReplyEmpty| { let queued = if replacement_authorized { self.core.queue_rename_authorized( ino, @@ -2031,6 +2042,14 @@ impl ProtonFs { reply.error(e); } } + }; + if rename_needs_queue( + self.core.is_online(), + is_local_uid(&new_parent_uid), + newparent != parent, + newname != name, + ) { + queue_it(reply); return; } // Rename first if both halves change. Moving first makes the encrypted @@ -2051,6 +2070,10 @@ impl ProtonFs { if newname != name && let Err(e) = self.core.rename_remote(&uid, newname) { + if self.core.lost_link(&e, "rename") { + queue_it(reply); + return; + } error!(%uid, error = %e, "rename failed"); self.core.restore_replaced(victim.as_ref(), newname); reply.error(Errno::EIO); @@ -2061,8 +2084,7 @@ impl ProtonFs { let moved = loop { match self .core - .rt - .block_on(self.core.client.move_node(&uid, &new_parent_uid)) + .block_on_bounded(self.core.client.move_node(&uid, &new_parent_uid)) { Ok(()) => break Ok(()), Err(e) @@ -2081,6 +2103,10 @@ impl ProtonFs { } }; if let Err(e) = moved { + if self.core.lost_link(&e, "move") { + queue_it(reply); + return; + } error!(%uid, attempts, error = %e, "move after rename failed"); // The rename half landed in the source directory. let mut state = self.core.state(); @@ -2158,12 +2184,14 @@ impl ProtonFs { } // Offline: queue it. Trashing is the one mutation a user expects to work // regardless — the file is gone from their point of view the moment the - // command returns (offline.md Phase 3b). - if !self.core.online.load(Ordering::Relaxed) { - match self.core.queue_trash(&uid, name) { - Ok(()) => reply.ok(), - Err(e) => reply.error(e), - } + // command returns (offline.md Phase 3b). The same when the network goes + // away under the remote call below. + let queue_it = |reply: ReplyEmpty| match self.core.queue_trash(&uid, name) { + Ok(()) => reply.ok(), + Err(e) => reply.error(e), + }; + if !self.core.is_online() { + queue_it(reply); return; } if let Err(e) = self.core.require_uid_writable(&uid) { @@ -2172,10 +2200,13 @@ impl ProtonFs { } if let Err(e) = self .core - .rt - .block_on(self.core.client.trash_nodes(std::slice::from_ref(&uid))) + .block_on_bounded(self.core.client.trash_nodes(std::slice::from_ref(&uid))) .and_then(batch::into_unit) { + if self.core.lost_link(&e, "trash") { + queue_it(reply); + return; + } error!(%uid, error = %e, "trash failed"); self.core .log_activity(ActivityKind::Trash, name, e.to_string(), false); diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 94f8a4f..0eeb407 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -88,6 +88,7 @@ mod diagnostics; mod drain; mod events; mod filesystem; +mod link; pub use filesystem::ProtonFs; mod mount; mod pause; @@ -155,10 +156,6 @@ const MAX_NAME_LEN: u32 = 255; /// How often the background task polls the remote event cursor. const POLL_INTERVAL: Duration = Duration::from_secs(10); -/// First and longest delay between probes for the network coming back after an -/// offline mount (offline.md Phase 1). Doubles from min to max: a laptop shut in -/// a bag is the common case, so the steady state must be cheap, while a brief -/// blip should still recover in seconds. /// Retry backoff for a queued upload, doubling per attempt between these. The /// floor is short because the common failure is a brief network blip; the /// ceiling keeps a persistently failing op from spinning. @@ -222,6 +219,10 @@ const DRAIN_WORKERS: usize = 3; /// path to go and fetch them from. const STAGING_ORPHAN_RETAIN: Duration = Duration::from_secs(30 * 24 * 60 * 60); +/// First and longest delay between probes for the network coming back after an +/// offline mount (offline.md Phase 1). Doubles from min to max: a laptop shut in +/// a bag is the common case, so the steady state must be cheap, while a brief +/// blip should still recover in seconds. const ONLINE_PROBE_MIN: Duration = Duration::from_secs(5); const ONLINE_PROBE_MAX: Duration = Duration::from_secs(300); /// How long the persisted photos timeline stays good before a page request @@ -533,6 +534,9 @@ struct Core { /// `Response::Status` so the UI can say so rather than leaving the user to /// infer it from a wall of EIO. online: Arc, + /// When the link went down and a condvar for its coming back; the other + /// half of `online` (see [`link`]). + link: Arc, /// When the user paused syncing, the Unix second it resumes by itself; /// [`pause::PAUSED_INDEFINITELY`] for "until resumed", `0` when not paused. /// Persisted, so a pause survives a restart (see [`Core::sync_paused`]). @@ -646,6 +650,9 @@ struct Core { /// storage moves slowly enough that a minute-old number is a better answer /// than either a stall or the zeroes the default implementation returns. quota: Arc>>, + /// Set while a `statfs` is refreshing [`Core::quota`], so the others serve + /// the last figures instead of each holding a worker on the same request. + quota_refreshing: Arc, /// Database maintenance that has been moved off the request thread. /// /// `VACUUM` and a deep `integrity_check` both hold the single connection for @@ -1567,7 +1574,7 @@ impl Core { /// follows an earlier rename of the same node. See [`retry_stale_rename`]. pub(crate) fn rename_remote(&self, uid: &NodeUid, name: &str) -> Result<(), ProtonError> { retry_stale_rename( - || self.rt.block_on(self.client.rename_node(uid, name, None)), + || self.block_on_bounded(self.client.rename_node(uid, name, None)), || std::thread::sleep(RENAME_RETRY_DELAY), ) } @@ -1964,47 +1971,6 @@ impl Core { } } - /// Poll for the API becoming reachable again after an offline mount, then - /// flip `online` and refresh the root (offline.md Phase 1). Runs on its own - /// thread and returns once we are back online: nothing sets `online` false - /// again, because a mount that has been online once keeps its live event - /// sync, which does its own retrying. - /// - /// Backs off to [`ONLINE_PROBE_MAX`] rather than hammering a fixed interval — - /// a laptop can sit offline for days, and each probe is a real API round trip. - fn run_online_probe(&self) { - let mut delay = ONLINE_PROBE_MIN; - loop { - if !self.shutdown.sleep(delay) { - return; - } - match self.rt.block_on(self.client.get_my_files_folder()) { - Ok(root) => { - { - let mut st = self.state(); - if let Some(e) = st.entries.get_mut(&ROOT_INO) { - e.node = root.clone(); - } - } - if let Err(e) = self.db.upsert_node(&root) { - warn!(error = %e, "refresh root after reconnect failed"); - } - self.online.store(true, Ordering::Relaxed); - // Anything written while offline is queued and waiting on - // exactly this. - self.wake_drain(); - info!("back online"); - self.repair_primary_share_id(&root.uid); - return; - } - Err(e) => { - debug!(error = %e, ?delay, "online probe failed; still offline"); - delay = (delay * 2).min(ONLINE_PROBE_MAX); - } - } - } - } - /// Whether `ino`'s listing is already in memory, i.e. whether /// [`Core::ensure_children`] would return without touching the network. /// Lets a handler decide between answering inline and handing off to a @@ -2251,6 +2217,98 @@ impl Core { Ok(()) } + /// Serve a listing without the network: the one already in memory if any, + /// else whatever children the DB still knows. The fallback only answers + /// until the link is back, when [`Core::relist_offline_listings`] drops it: + /// a listing dropped as stale may lack a child another device made, and a + /// create over that name lands as a conflict copy rather than an `EIO` now. + fn adopt_stale_listing( + &self, + ino: u64, + folder_uid: &NodeUid, + cached: bool, + primary_root: bool, + ) -> Result<(), Errno> { + if cached { + return Ok(()); + } + let nodes = self.db.known_children(folder_uid).map_err(|e| { + warn!(%folder_uid, error = %e, "db known_children failed"); + Errno::EIO + })?; + debug!(%folder_uid, count = nodes.len(), "offline: listing a folder from what the DB knows"); + self.link.stale_listings.lock().insert(ino); + self.adopt_db_listing(ino, folder_uid, nodes, primary_root) + } + + /// Relist every folder [`Core::adopt_stale_listing`] served while offline. + pub(crate) fn relist_offline_listings(&self) { + let stale = std::mem::take(&mut *self.link.stale_listings.lock()); + if stale.is_empty() { + return; + } + let mut st = self.state(); + for &ino in &stale { + st.invalidate_listing(ino); + } + drop(st); + if let Some(notifier) = self.notifier.get() { + for ino in stale { + let _ = notifier.inval_inode(INodeNo(ino), 0, 0); + } + } + } + + /// Install `nodes`, read from the DB, as `ino`'s listing. + fn adopt_db_listing( + &self, + ino: u64, + folder_uid: &NodeUid, + mut nodes: Vec, + primary_root: bool, + ) -> Result<(), Errno> { + let folder_uid = folder_uid.clone(); + if primary_root { + nodes.retain(|node| !is_virtual_uid(&node.uid)); + } + // Before the lock: a DB row carries the size the server last + // sealed, which a queued write is ahead of (B11). + self.stamp_pending_sizes(&mut nodes); + let hidden = self.hidden.lock().clone(); + let mut st = self.state(); + if st.children.contains_key(&ino) { + return Ok(()); + } + let mut child_inos = Vec::with_capacity(nodes.len()); + let mut needs_size = Vec::new(); + for node in nodes { + if !node_visible(&node, &folder_uid, &hidden) { + continue; + } + if matches!( + &node.kind, + NodeKind::File { + claimed_size: None, + .. + } + ) { + needs_size.push(node.uid.clone()); + } + child_inos.push(st.intern_from_db(ino, node)); + } + st.children.insert(ino, child_inos); + let root_snapshot = primary_root.then(|| RootListingSnapshot::capture(&st, ino).unwrap()); + drop(st); + self.flush_access_changes(); + if let Some(snapshot) = root_snapshot { + self.reconcile_virtual_root_dentry(ino, snapshot)?; + } + // Rows persisted from a cheap enumeration whose upgrade never + // ran (a restart in between, say) still owe their real sizes. + self.spawn_size_upgrade(ino, needs_size); + Ok(()) + } + /// Enumerate `ino`'s children from the remote and cache them. No-op if the /// directory has already been listed. Network I/O happens without the lock /// held so concurrent metadata reads aren't blocked behind a fetch. @@ -2291,61 +2349,29 @@ impl Core { self.db.children_if_listed(&folder_uid) }; match cached_nodes { - Ok(Some(mut nodes)) => { - if primary_root { - nodes.retain(|node| !is_virtual_uid(&node.uid)); - } - // Before the lock: a DB row carries the size the server last - // sealed, which a queued write is ahead of (B11). - self.stamp_pending_sizes(&mut nodes); - let hidden = self.hidden.lock().clone(); - let mut st = self.state(); - if st.children.contains_key(&ino) { - return Ok(()); - } - let mut child_inos = Vec::with_capacity(nodes.len()); - let mut needs_size = Vec::new(); - for node in nodes { - if !node_visible(&node, &folder_uid, &hidden) { - continue; - } - if matches!( - &node.kind, - NodeKind::File { - claimed_size: None, - .. - } - ) { - needs_size.push(node.uid.clone()); - } - child_inos.push(st.intern_from_db(ino, node)); - } - st.children.insert(ino, child_inos); - let root_snapshot = - primary_root.then(|| RootListingSnapshot::capture(&st, ino).unwrap()); - drop(st); - self.flush_access_changes(); - if let Some(snapshot) = root_snapshot { - self.reconcile_virtual_root_dentry(ino, snapshot)?; - } - // Rows persisted from a cheap enumeration whose upgrade never - // ran (a restart in between, say) still owe their real sizes. - self.spawn_size_upgrade(ino, needs_size); - return Ok(()); - } + Ok(Some(nodes)) => return self.adopt_db_listing(ino, &folder_uid, nodes, primary_root), Ok(None) => {} Err(e) => warn!(%folder_uid, error = %e, "db children_if_listed failed"), } + if !self.is_online() { + return self.adopt_stale_listing(ino, &folder_uid, cached, primary_root); + } let refresh_generation = refresh_foreign.then(|| self.shared_generation.load(Ordering::SeqCst)); - let uids = self + let uids = match self .rt .block_on(self.client.enumerate_folder_children_node_uids(&folder_uid)) - .map_err(|e| { + { + Ok(uids) => uids, + Err(e) if self.lost_link(&e, "list folder") => { + return self.adopt_stale_listing(ino, &folder_uid, cached, primary_root); + } + Err(e) => { error!(%folder_uid, error = %e, "enumerate folder children failed"); - Errno::EIO - })?; + return Err(Errno::EIO); + } + }; // Cheap enumeration: `Light` skips unlocking each *file's* node key, // which is an S2K derivation per file and was ~74% of the cost of a cold // listing (B12 — measured with `perf`, 64% of cycles in SHA-256 alone). @@ -2355,13 +2381,16 @@ impl Core { // The price is that files come back without a `claimed_size`, so // `node_size` falls back to the *ciphertext* size until // `spawn_size_upgrade` below fills the real one in. - let mut nodes = self - .rt - .block_on(self.client.enumerate_nodes_light(&uids)) - .map_err(|e| { + let mut nodes = match self.rt.block_on(self.client.enumerate_nodes_light(&uids)) { + Ok(nodes) => nodes, + Err(e) if self.lost_link(&e, "list folder") => { + return self.adopt_stale_listing(ino, &folder_uid, cached, primary_root); + } + Err(e) => { error!(%folder_uid, error = %e, "enumerate nodes failed"); - Errno::EIO - })?; + return Err(Errno::EIO); + } + }; // Same as the DB path above: the remote's size for a file with a write // still queued is the pre-write one (B11). self.stamp_pending_sizes(&mut nodes); @@ -2645,9 +2674,15 @@ impl Core { /// wait times out, so the read cannot rely on it (bugs.md B100). fn resolve_provisional_size(&self, ino: u64, uid: &NodeUid) -> Option { let parent = self.state().entries.get(&ino)?.parent; - let result = self - .rt - .block_on(self.client.enumerate_nodes(std::slice::from_ref(uid))); + // Without a network the answer is `None` either way; the bound keeps a + // dead link from holding the open or read for the SDK's two minutes. + if !self.is_online() { + return None; + } + let result = self.block_on_bounded(self.client.enumerate_nodes(std::slice::from_ref(uid))); + if let Err(e) = &result { + self.lost_link(e, "settle a provisional size"); + } self.apply_size_upgrade(parent, result); let st = self.state(); match &st.entries.get(&ino)?.node.kind { @@ -2874,7 +2909,9 @@ impl Core { Ok(Some(node)) => Ok(node), Ok(None) => Err(Errno::ENOENT), Err(e) => { - error!(%uid, error = %e, "enumerate node failed"); + if !self.lost_link(&e, "fetch node") { + error!(%uid, error = %e, "enumerate node failed"); + } Err(Errno::EIO) } } @@ -2884,10 +2921,7 @@ impl Core { /// resolving a conflict turns on *why* a call failed, and "the node is not /// there" (`Ok(None)`) is a different outcome from "we could not ask". fn fetch_node_remote(&self, uid: &NodeUid) -> Result, ProtonError> { - match self - .rt - .block_on(self.client.enumerate_nodes(std::slice::from_ref(uid))) - { + match self.block_on_bounded(self.client.enumerate_nodes(std::slice::from_ref(uid))) { Ok(nodes) => Ok(nodes.into_iter().next()), // An unknown uid is reported either as an empty result or as an // outright refusal, depending on the endpoint. @@ -4487,12 +4521,19 @@ impl Core { debug!(%uid, name, "replaced a node whose create was still queued"); return Ok(()); } + // Offline, or the network goes away under the call: queue the trash. + // The rename that follows queues as well, behind it. + if !self.is_online() { + return self.queue_trash(uid, name); + } self.require_uid_writable(uid)?; if let Err(e) = self - .rt - .block_on(self.client.trash_nodes(std::slice::from_ref(uid))) + .block_on_bounded(self.client.trash_nodes(std::slice::from_ref(uid))) .and_then(batch::into_unit) { + if self.lost_link(&e, "trash a replaced node") { + return self.queue_trash(uid, name); + } error!(%uid, name, error = %e, "trashing the node a rename replaces failed"); self.log_activity(ActivityKind::Trash, name, e.to_string(), false); return Err(Errno::EIO); @@ -8216,6 +8257,21 @@ mod tests { } } + #[test] + fn a_folder_lists_from_the_db_while_the_link_is_down() { + // A listing dropped as stale mid-run, then a drop: every create in that + // folder looks the name up first, and answered EIO for want of it. + let ensure = function_source(include_str!("lib.rs"), "fn ensure_children("); + assert_before( + ensure, + "if !self.is_online()", + "enumerate_folder_children_node_uids", + ); + assert_eq!(ensure.matches("adopt_stale_listing(").count(), 3); + let online = function_source(include_str!("link.rs"), "fn mark_online("); + assert!(online.contains("self.relist_offline_listings();")); + } + #[test] fn a_remote_read_gives_up_at_its_deadline() { // The SDK bounds requests, not reads; a stalled link held one `read` @@ -8224,11 +8280,12 @@ mod tests { assert_before( remote, "let fetch = async", - "tokio::time::timeout(READ_FETCH_TIMEOUT, fetch)", + "tokio::time::timeout(budget, fetch)", ); + assert!(remote.contains("let deadline = Instant::now() + READ_FETCH_TIMEOUT;")); // Built inside the runtime: a FUSE worker has no reactor, and a timer // built there panicked every remote read. - assert!(remote.contains("async { tokio::time::timeout(READ_FETCH_TIMEOUT, fetch).await }")); + assert!(remote.contains("async { tokio::time::timeout(budget, fetch).await }")); } #[test] diff --git a/crates/pdfs-fuse/src/link.rs b/crates/pdfs-fuse/src/link.rs new file mode 100644 index 0000000..d08c1ba --- /dev/null +++ b/crates/pdfs-fuse/src/link.rs @@ -0,0 +1,315 @@ +//! Noticing that the network went away mid-session, and that it came back. +//! +//! The mount used to learn it was offline only at startup: a daemon that had +//! connected once kept `online` true for the rest of its life, so a Wi-Fi drop +//! turned every create, mkdir, delete and rename into a two-minute wait for an +//! `EIO`, while the queue that exists for exactly this sat unused. Now any +//! remote call a handler waits on reports a failure to reach Proton through +//! [`Core::lost_link`], which flips the mount offline; the handler then takes its +//! queued path, and the probe thread here flips it back when Proton answers. + +use std::future::Future; +use std::sync::atomic::Ordering; +use std::time::{Duration, Instant}; + +use parking_lot::{Condvar, Mutex}; +use proton_drive_rs::proton_sdk::api::ResponseCode; +use proton_drive_rs::proton_sdk::error::{ProtonApiError, ProtonError}; +use tracing::{debug, info, warn}; + +use crate::{Core, ONLINE_PROBE_MAX, ONLINE_PROBE_MIN, ROOT_INO}; + +/// How long a remote call a FUSE handler is waiting on may take before the link +/// is treated as down. +/// +/// The SDK's own budget for one request is four attempts of up to 30 seconds +/// each, so a dead link held the caller for about two minutes. A metadata call +/// that has had no answer in 20 seconds is not going to be one the user wants to +/// keep waiting for; queueing the change is the better answer. +pub(crate) const INTERACTIVE_CALL_TIMEOUT: Duration = Duration::from_secs(20); + +/// How long one probe may wait for an answer. A probe is a single small request, +/// and a probe stuck in the SDK's retry budget would delay noticing the link is +/// back by as long as it waits. +const PROBE_TIMEOUT: Duration = Duration::from_secs(10); + +/// First delay between probes after a drop seen mid-session, and the ceiling it +/// doubles to while the outage is young. A blip should end in seconds, not in +/// the minutes [`ONLINE_PROBE_MAX`] allows. +const PROBE_AFTER_DROP: Duration = Duration::from_secs(2); +const PROBE_YOUNG_MAX: Duration = Duration::from_secs(30); +/// How long an outage counts as young. Past it the probe backs off to +/// [`ONLINE_PROBE_MAX`], because a laptop can sit offline for days and each +/// probe is a real API round trip. +const YOUNG_OUTAGE: Duration = Duration::from_secs(10 * 60); + +/// How often the idle probe thread checks for teardown while the link is up. +const IDLE_CHECK: Duration = Duration::from_secs(1); + +/// The shared half of the online flag: when the link went down, and a condvar +/// that fires on every change so waiters need not poll. +#[derive(Default)] +pub(crate) struct Link { + lost_at: Mutex>, + changed: Condvar, + /// Folders listed from the DB alone while offline, to relist once back. + pub(crate) stale_listings: Mutex>, +} + +/// Whether a failed call failed because Proton could not be reached, rather +/// than because it answered and refused. +/// +/// A transport error is the network. So is a request timeout: the SDK reports +/// Proton's own "try again" codes that way, and [`no_answer`] reports our +/// deadline that way. Anything else Proton actually said, and says again on a +/// retry, so it must reach the caller rather than be queued. +pub(crate) fn is_network_error(e: &ProtonError) -> bool { + match e { + ProtonError::Transport(_) => true, + ProtonError::Api(api) => { + matches!( + api.code, + ResponseCode::RequestTimeout | ResponseCode::Timeout + ) + } + _ => false, + } +} + +/// The error a call reports when it runs past `after` without an answer. +pub(crate) fn no_answer(after: Duration) -> ProtonError { + ProtonError::Api(ProtonApiError { + code: ResponseCode::RequestTimeout, + http_status: 408, + message: format!("no answer within {}s", after.as_secs()), + details: None, + }) +} + +/// Run a remote call, giving up after [`INTERACTIVE_CALL_TIMEOUT`]. +pub(crate) async fn bounded( + call: impl Future>, +) -> Result { + match tokio::time::timeout(INTERACTIVE_CALL_TIMEOUT, call).await { + Ok(result) => result, + Err(_) => Err(no_answer(INTERACTIVE_CALL_TIMEOUT)), + } +} + +/// The next delay between probes, given the current one and how long the +/// outage has lasted. +fn next_probe_delay(delay: Duration, outage: Duration) -> Duration { + let cap = if outage < YOUNG_OUTAGE { + PROBE_YOUNG_MAX + } else { + ONLINE_PROBE_MAX + }; + (delay * 2).min(cap) +} + +impl Core { + /// Whether the mount currently believes Proton is reachable. + pub(crate) fn is_online(&self) -> bool { + self.online.load(Ordering::Relaxed) + } + + /// Run a remote call a caller is waiting on, giving up after + /// [`INTERACTIVE_CALL_TIMEOUT`]. The timer is built inside the runtime: a + /// FUSE worker has no reactor of its own. + pub(crate) fn block_on_bounded( + &self, + call: impl Future>, + ) -> Result { + self.rt.block_on(bounded(call)) + } + + /// Report a failed remote call. Returns whether it failed for want of a + /// network, in which case the mount is now offline and the caller should + /// take its queued path instead of failing. + pub(crate) fn lost_link(&self, e: &ProtonError, what: &str) -> bool { + if !is_network_error(e) { + return false; + } + self.mark_offline(what, e); + true + } + + /// Flip the mount offline and wake the probe. Only the first caller of an + /// outage logs it; the rest are the same outage. + pub(crate) fn mark_offline(&self, what: &str, error: &dyn std::fmt::Display) { + let mut lost_at = self.link.lost_at.lock(); + if self.online.swap(false, Ordering::Relaxed) { + *lost_at = Some(Instant::now()); + warn!(during = what, %error, + "lost the connection to Proton; queueing changes until it returns"); + self.link.changed.notify_all(); + } + } + + /// Flip the mount back online, wake everything waiting for it, and let the + /// queue drain. + fn mark_online(&self) { + { + let mut lost_at = self.link.lost_at.lock(); + self.online.store(true, Ordering::Relaxed); + match lost_at.take() { + Some(since) => { + info!( + offline_secs = since.elapsed().as_secs(), + "connection to Proton is back" + ) + } + None => info!("back online"), + } + self.link.changed.notify_all(); + } + self.relist_offline_listings(); + // Anything written while offline is queued and waiting on exactly this, + // including ops that failed on the way down and are sitting out a + // backoff sized for a server that keeps refusing. + if let Err(e) = self.db.retry_failed_ops_now(crate::now_millis()) { + warn!(error = %e, "moving backed-off ops up after a reconnect failed"); + } + self.wake_drain(); + } + + /// Wait up to `budget` for the link to come back. Returns whether it is up. + pub(crate) fn wait_online(&self, budget: Duration) -> bool { + let deadline = Instant::now() + budget; + let mut lost_at = self.link.lost_at.lock(); + while !self.is_online() { + if self.shutdown.is_stopping() + || self + .link + .changed + .wait_until(&mut lost_at, deadline) + .timed_out() + { + return self.is_online(); + } + } + true + } + + /// Wait out `interval` before a background poll of Proton, and while the + /// link is down, until it is back. Returns as soon as it comes back, so the + /// poll catches up on what changed during the outage at once instead of + /// one interval later. A poll into a dead link would only wait out the + /// SDK's retries and fail. + pub(crate) async fn next_poll(&self, interval: Duration) { + let due = Instant::now() + interval; + let mut was_offline = false; + loop { + if !self.is_online() { + was_offline = true; + } else if was_offline || Instant::now() >= due { + return; + } + tokio::time::sleep(IDLE_CHECK).await; + } + } + + /// Watch the link for the life of the mount: sleep while it is up, and + /// probe it back while it is down (offline.md Phase 1). + /// + /// A mount that started from the cache is down from the first moment; one + /// that started online goes down when [`Core::mark_offline`] says so. + pub(crate) fn run_online_probe(&self) { + // Only an offline start never had a chance to repair its share id. + let mut repair_share_id = !self.is_online(); + loop { + { + let mut lost_at = self.link.lost_at.lock(); + while self.is_online() { + if self.shutdown.is_stopping() { + return; + } + self.link.changed.wait_for(&mut lost_at, IDLE_CHECK); + } + } + if !self.probe_until_online() { + return; + } + if std::mem::take(&mut repair_share_id) { + self.repair_primary_share_id(&self.primary_root_uid); + } + } + } + + /// Probe until Proton answers, then flip the mount online. Returns `false` + /// if the daemon is stopping. + fn probe_until_online(&self) -> bool { + let started = self.link.lost_at.lock().unwrap_or_else(Instant::now); + let mut delay = if started.elapsed() < YOUNG_OUTAGE { + PROBE_AFTER_DROP + } else { + ONLINE_PROBE_MIN + }; + loop { + if !self.shutdown.sleep(delay) { + return false; + } + let probe = self.rt.block_on(async { + match tokio::time::timeout(PROBE_TIMEOUT, self.client.get_my_files_folder()).await { + Ok(result) => result, + Err(_) => Err(no_answer(PROBE_TIMEOUT)), + } + }); + match probe { + Ok(root) => { + { + let mut st = self.state(); + if let Some(e) = st.entries.get_mut(&ROOT_INO) { + e.node = root.clone(); + } + } + if let Err(e) = self.db.upsert_node(&root) { + warn!(error = %e, "refresh root after reconnect failed"); + } + self.mark_online(); + return true; + } + Err(e) => { + debug!(error = %e, ?delay, "online probe failed; still offline"); + delay = next_probe_delay(delay, started.elapsed()); + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn api(code: ResponseCode) -> ProtonError { + ProtonError::Api(ProtonApiError { + code, + http_status: 422, + message: String::new(), + details: None, + }) + } + + #[test] + fn only_an_unanswered_call_counts_as_a_lost_link() { + assert!(is_network_error(&no_answer(INTERACTIVE_CALL_TIMEOUT))); + assert!(is_network_error(&api(ResponseCode::Timeout))); + assert!(!is_network_error(&api(ResponseCode::AlreadyExists))); + assert!(!is_network_error(&api(ResponseCode::InvalidRequirements))); + assert!(!is_network_error(&ProtonError::invalid_operation("bug"))); + } + + #[test] + fn a_young_outage_is_probed_often_and_an_old_one_rarely() { + let young = Duration::from_secs(60); + let old = YOUNG_OUTAGE + Duration::from_secs(1); + assert_eq!( + next_probe_delay(PROBE_AFTER_DROP, young), + 2 * PROBE_AFTER_DROP + ); + assert_eq!(next_probe_delay(PROBE_YOUNG_MAX, young), PROBE_YOUNG_MAX); + assert_eq!(next_probe_delay(PROBE_YOUNG_MAX, old), 2 * PROBE_YOUNG_MAX); + assert_eq!(next_probe_delay(ONLINE_PROBE_MAX, old), ONLINE_PROBE_MAX); + } +} diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 5b159f5..038f503 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -383,6 +383,7 @@ pub fn mount( shared_generation: Arc::new(AtomicU64::new(0)), shared_refresh_deadlines: Arc::new(Mutex::new(SharedRefreshDeadlines::default())), online: Arc::new(AtomicBool::new(online)), + link: Arc::new(Default::default()), sync_paused_until: Arc::new(AtomicI64::new(paused_until)), pending: Arc::new(Mutex::new(HashMap::new())), hidden: Arc::new(Mutex::new(HashSet::new())), @@ -411,6 +412,7 @@ pub fn mount( thumbnail_build_cancel: Arc::new(tokio::sync::Notify::new()), thumbnail_misses: Arc::new(Mutex::new(Default::default())), quota: Arc::new(Mutex::new(None)), + quota_refreshing: Arc::new(AtomicBool::new(false)), size_upgrades: Arc::new(Mutex::new(HashMap::new())), size_waiters: Arc::new(Default::default()), notifier: Arc::new(OnceLock::new()), @@ -484,9 +486,10 @@ pub fn mount( ); } - // Mounted from the cache: watch for the network coming back so the mount can - // stop being read-only-ish without the user restarting the daemon. - if !online { + // Watch the link for the life of the mount: a mount from the cache waits for + // the network to come back, and an online one can lose it at any moment and + // must notice it coming back just the same. + { let core = core.clone(); workers.push( std::thread::Builder::new() diff --git a/crates/pdfs-fuse/src/reads.rs b/crates/pdfs-fuse/src/reads.rs index 0a3313d..4d27b38 100644 --- a/crates/pdfs-fuse/src/reads.rs +++ b/crates/pdfs-fuse/src/reads.rs @@ -45,6 +45,12 @@ pub(super) const PREFETCH_BUDGET: usize = 8; /// caller can retry beats a process stuck in `D` state. pub(super) const READ_FETCH_TIMEOUT: Duration = Duration::from_secs(120); +/// How long a read of blocks nothing has cached waits for a lost link to come +/// back before failing. Long enough to ride out a Wi-Fi blip, short enough +/// that a program reading while the laptop is really offline gets its `EIO` +/// before the user gives up on it. +const READ_OFFLINE_WAIT: Duration = Duration::from_secs(30); + /// Cap on tracked sequential streams. Only reads in progress matter, so the map /// is cleared wholesale rather than aged — losing the ramp costs one read. const MAX_PREFETCH_STREAMS: usize = 256; @@ -373,6 +379,7 @@ impl Core { let result = match self.client.open_revision(uid).await { Ok(reader) => Ok(Arc::new(reader)), Err(e) => { + self.lost_link(&e, "open a revision"); warn!(%uid, error = %e, "open_revision failed"); Err(Errno::EIO) } @@ -658,6 +665,7 @@ impl Core { .store_block_geometry(uid, mtime, fsize, reader.block_sizes()); } let bytes = reader.read_at(span.start, span.len).await.map_err(|e| { + self.lost_link(&e, "read a block"); warn!(%uid, start = span.start, len = span.len, error = %e, "block read failed"); Errno::EIO })?; @@ -948,43 +956,67 @@ impl Core { } if !misses.is_empty() { - let fetch = async { - // Resolve the file's keys and block table once, then read every - // missing block through the shared reader. Previously each block - // called `download_range`, which redid that resolution per block. - let reader = self.revision_reader(uid, mtime, fsize).await?; + // One deadline for the whole read, including any wait for a lost + // link. A fetch that failed because the link went down is tried + // once more when it is back, so a blip is a slow read, not an `EIO`. + let deadline = Instant::now() + READ_FETCH_TIMEOUT; + let mut retried = false; + let fetched = loop { + if !self.is_online() { + let wait = + READ_OFFLINE_WAIT.min(deadline.saturating_duration_since(Instant::now())); + if !self.wait_online(wait) { + debug!(%uid, offset, len, "read of uncached blocks while offline; failing it"); + return Err(Errno::EIO); + } + } + let fetch = async { + // Resolve the file's keys and block table once, then read every + // missing block through the shared reader. Previously each block + // called `download_range`, which redid that resolution per block. + let reader = self.revision_reader(uid, mtime, fsize).await?; - let mut set = tokio::task::JoinSet::new(); - for &span in &misses { - let (core, reader, uid) = (self.clone(), reader.clone(), uid.clone()); - set.spawn(async move { - core.fetch_block(&reader, &uid, mtime, fsize, span, cache_blocks) - .await - .map(|bytes| (span.idx, bytes)) - }); + let mut set = tokio::task::JoinSet::new(); + for &span in &misses { + let (core, reader, uid) = (self.clone(), reader.clone(), uid.clone()); + set.spawn(async move { + core.fetch_block(&reader, &uid, mtime, fsize, span, cache_blocks) + .await + .map(|bytes| (span.idx, bytes)) + }); + } + let mut out = Vec::with_capacity(misses.len()); + while let Some(joined) = set.join_next().await { + // A join error means the task panicked; surface it as EIO. + out.push(joined.map_err(|_| Errno::EIO)??); + } + Ok::<_, Errno>(out) + }; + // Dropping the fetch on the deadline aborts the block tasks with + // it; blocks that already landed stay cached for the retry. The + // timer is built inside the runtime: a FUSE worker has no reactor + // of its own, and building it out here panicked every remote read. + let budget = deadline.saturating_duration_since(Instant::now()); + match self + .rt + .block_on(async { tokio::time::timeout(budget, fetch).await }) + { + Ok(Ok(fetched)) => break fetched, + // The fetch saw the link go down; the top of the loop waits + // for it within what is left of the deadline. + Ok(Err(_)) if !retried && !self.is_online() => retried = true, + Ok(Err(e)) => return Err(e), + Err(_) => { + warn!( + %uid, offset, len, secs = READ_FETCH_TIMEOUT.as_secs(), + "read got no answer from the network in time; failing it" + ); + self.evict_reader(uid); + self.mark_offline("read", &crate::link::no_answer(READ_FETCH_TIMEOUT)); + return Err(Errno::EIO); + } } - let mut out = Vec::with_capacity(misses.len()); - while let Some(joined) = set.join_next().await { - // A join error means the task panicked; surface it as EIO. - out.push(joined.map_err(|_| Errno::EIO)??); - } - Ok::<_, Errno>(out) }; - // Dropping the fetch on the deadline aborts the block tasks with it; - // blocks that already landed stay cached for the retry. The timer is - // built inside the runtime: a FUSE worker has no reactor of its own, - // and building it out here panicked every remote read. - let fetched = self - .rt - .block_on(async { tokio::time::timeout(READ_FETCH_TIMEOUT, fetch).await }) - .unwrap_or_else(|_| { - warn!( - %uid, offset, len, secs = READ_FETCH_TIMEOUT.as_secs(), - "read got no answer from the network in time; failing it" - ); - self.evict_reader(uid); - Err(Errno::EIO) - })?; for (bidx, bytes) in fetched { blocks[(bidx - first) as usize] = Some(bytes); } diff --git a/crates/pdfs-fuse/src/sharing.rs b/crates/pdfs-fuse/src/sharing.rs index f058f45..65b7df2 100644 --- a/crates/pdfs-fuse/src/sharing.rs +++ b/crates/pdfs-fuse/src/sharing.rs @@ -610,10 +610,10 @@ impl Core { /// Proton products (not Drive-only). A remote round-trip; nothing here is /// cached. pub(crate) fn account_quota(&self) -> CoreResult<(i64, i64)> { - let q = self - .rt - .block_on(self.client.quota()) - .map_err(|e| CoreError::from_api(&e, "account quota"))?; + let q = self.block_on_bounded(self.client.quota()).map_err(|e| { + self.lost_link(&e, "account quota"); + CoreError::from_api(&e, "account quota") + })?; *self.quota.lock() = Some((std::time::Instant::now(), q.max_space, q.used_space)); Ok((q.max_space, q.used_space)) } @@ -632,11 +632,22 @@ impl Core { { return Some((max_space, used_space)); } - match self.account_quota() { + let last = || self.quota.lock().map(|(_, max, used)| (max, used)); + if !self.is_online() + || self + .quota_refreshing + .swap(true, std::sync::atomic::Ordering::AcqRel) + { + return last(); + } + let refreshed = self.account_quota(); + self.quota_refreshing + .store(false, std::sync::atomic::Ordering::Release); + match refreshed { Ok(q) => Some(q), Err(error) => { debug!(%error, "account quota refresh failed; serving the last known figures"); - self.quota.lock().map(|(_, max, used)| (max, used)) + last() } } }