diff --git a/Cargo.lock b/Cargo.lock index adf03a8..9e155cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3515,6 +3515,7 @@ dependencies = [ name = "pdfs-fuse" version = "2.8.2" dependencies = [ + "async-trait", "base64 0.23.1", "fuser", "futures", diff --git a/crates/pdfs-fuse/Cargo.toml b/crates/pdfs-fuse/Cargo.toml index 61868fa..bf923ba 100644 --- a/crates/pdfs-fuse/Cargo.toml +++ b/crates/pdfs-fuse/Cargo.toml @@ -15,6 +15,7 @@ zip.workspace = true sha1.workspace = true proton-drive-rs.workspace = true futures.workspace = true +async-trait.workspace = true tokio.workspace = true tracing.workspace = true fuser.workspace = true diff --git a/crates/pdfs-fuse/src/background.rs b/crates/pdfs-fuse/src/background.rs index 048c520..ad734a0 100644 --- a/crates/pdfs-fuse/src/background.rs +++ b/crates/pdfs-fuse/src/background.rs @@ -323,7 +323,7 @@ fn flush_dirty_parents(core: &Core, dirty: &DirtyParents) { /// to be applied to every mounted inode space, and the registry that enumerates /// them lives on the `Core` (see [`apply_event`]). pub(super) async fn run_event_sync( - client: ProtonDriveClient, + client: Arc, scope: DriveEventScopeId, core: Core, ) { diff --git a/crates/pdfs-fuse/src/devices.rs b/crates/pdfs-fuse/src/devices.rs index bc94cd4..2579f28 100644 --- a/crates/pdfs-fuse/src/devices.rs +++ b/crates/pdfs-fuse/src/devices.rs @@ -363,14 +363,14 @@ impl Core { ) -> CoreResult> { let uids = self .rt - .block_on(self.client.enumerate_folder_children_node_uids(root_uid)) + .block_on(self.drive.enumerate_folder_children_node_uids(root_uid)) .map_err(|e| CoreError::from_api(&e, "list device root"))?; if uids.is_empty() { return Ok(None); } let nodes = self .rt - .block_on(self.client.enumerate_nodes(&uids)) + .block_on(self.drive.enumerate_nodes(&uids)) .map_err(|e| CoreError::from_api(&e, "resolve device root children"))?; Ok(nodes .into_iter() @@ -446,10 +446,7 @@ impl Core { } None => self .rt - .block_on( - self.client - .create_folder(&root_uid, &name, Some(now_secs())), - ) + .block_on(self.drive.create_folder(&root_uid, &name, Some(now_secs()))) .map_err(|e| CoreError::from_api(&e, &format!("create device folder {name}")))?, }; @@ -603,7 +600,7 @@ impl Core { && let Some(uid) = parse_uid(&folder.remote_uid) && let Err(e) = self .rt - .block_on(self.client.trash_nodes(&[uid])) + .block_on(self.drive.trash_nodes(&[uid])) .and_then(batch::into_unit) { warn!(id, error = %e, "delete remote device folder failed"); @@ -858,7 +855,7 @@ impl Core { })?; let root = self .rt - .block_on(self.client.enumerate_nodes(std::slice::from_ref(&root_uid))) + .block_on(self.drive.enumerate_nodes(std::slice::from_ref(&root_uid))) .map_err(|e| SwitchBlocked::Failed(format!("fetch remote root: {e}")))? .into_iter() .next() @@ -1017,7 +1014,7 @@ impl Core { }; let root = match self .rt - .block_on(self.client.enumerate_nodes(std::slice::from_ref(&root_uid))) + .block_on(self.drive.enumerate_nodes(std::slice::from_ref(&root_uid))) { Ok(v) => match v.into_iter().next() { Some(n) => n, diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index 4845381..9b8eaad 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -786,7 +786,7 @@ impl Core { self.rt .block_on(async { let mut outcomes = - std::pin::pin!(self.client.move_nodes_streaming(vec![item], parent.clone())); + std::pin::pin!(self.drive.move_nodes_streaming(vec![item], parent.clone())); match outcomes.next().await { Some(Ok((_, outcome))) => outcome, Some(Err(e)) => Err(e), @@ -808,7 +808,7 @@ impl Core { let name = op.name.clone().unwrap_or_else(|| op.uid.clone()); match self .rt - .block_on(self.client.trash_nodes(std::slice::from_ref(&uid))) + .block_on(self.drive.trash_nodes(std::slice::from_ref(&uid))) .and_then(batch::into_unit) { Ok(()) => {} @@ -965,8 +965,8 @@ impl Core { ) -> 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))?; + .block_on(self.drive.enumerate_folder_children_node_uids(parent))?; + let children = self.rt.block_on(self.drive.enumerate_nodes_light(&uids))?; let Some(twin) = children .into_iter() .find(|node| node.name == name && !node.trashed) @@ -1000,7 +1000,7 @@ impl Core { true => { Ok(self .rt - .block_on(self.client.create_folder(parent, name, Some(now_secs())))?) + .block_on(self.drive.create_folder(parent, name, Some(now_secs())))?) } false => self.upload_created_file(op, parent, name), } @@ -1015,7 +1015,7 @@ impl Core { name: &str, ) -> Result> { let Some(blob) = op.blob_path.as_deref() else { - return Ok(self.rt.block_on(self.client.upload_file( + return Ok(self.rt.block_on(self.drive.upload_file( parent, name, media_type_for(name), @@ -1035,7 +1035,7 @@ impl Core { .transfers .begin(name, op.uid.clone(), TransferDirection::Upload, meta.len); let reader = CountingReader::new(File::open(blob)?, &guard); - let uid = self.rt.block_on(self.client.upload_file_from( + let uid = self.rt.block_on(self.drive.upload_file_from( parent, name, media_type_for(name), @@ -1306,7 +1306,7 @@ impl Core { self.transfers .begin(&alt, meta.uid.clone(), TransferDirection::Upload, meta.len); let reader = CountingReader::new(File::open(blob)?, &guard); - self.rt.block_on(self.client.upload_file_from( + self.rt.block_on(self.drive.upload_file_from( &parent, &alt, media_type_for(&alt), @@ -1626,7 +1626,7 @@ impl Core { } 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( + let sent = self.rt.block_on(self.drive.upload_new_revision_from( &uid, reader, meta.len as i64, diff --git a/crates/pdfs-fuse/src/drive.rs b/crates/pdfs-fuse/src/drive.rs new file mode 100644 index 0000000..6eaff84 --- /dev/null +++ b/crates/pdfs-fuse/src/drive.rs @@ -0,0 +1,458 @@ +//! The calls the sync core makes to Drive, behind a trait. +//! +//! `Core` used to hold a concrete [`ProtonDriveClient`], so no test could make +//! Drive answer late, answer stale, or not answer at all, and the interleavings +//! behind most recent bugs could only be found on a real account over a slow +//! link (`docs/MILESTONE-3.0.0.md` §2, §8.1). [`DriveApi`] is the part of the +//! client the mount, the drain, the event feed and the mirror engine use; the +//! real client implements it by forwarding, and the simulation tests implement +//! it with an in-memory Drive that injects Drive's known faults. +//! +//! Sharing, Photos, devices, invitations and public links stay on the concrete +//! client (`Core::client`) until a test needs them faked. +//! +//! The method names match the client's own, so a call site reads the same +//! whichever it goes through. The client's generic methods (a reader or writer +//! of any type) cannot be part of an object-safe trait, so the trait takes them +//! as trait objects and `impl dyn DriveApi` puts the generic signatures back. + +use std::io::{Read, Write}; +use std::sync::Arc; + +use async_trait::async_trait; +use futures::StreamExt as _; +use futures::stream::BoxStream; +use proton_drive_rs::proton_sdk::error::Result; +use proton_drive_rs::proton_sdk::ids::{DriveEventId, NodeUid}; +use proton_drive_rs::{ + DriveEvent, DriveEventScopeId, Node, NodeMoveItem, ProtonDriveClient, RevisionReader, Thumbnail, +}; + +/// One outcome per node of a batch request; see [`ProtonDriveClient::trash_nodes`]. +pub(crate) type NodeOutcomes = Vec<(NodeUid, Result<()>)>; + +/// The outcomes of a streaming batch request, one item per node as it lands. +pub(crate) type OutcomeStream<'a> = BoxStream<'a, Result<(NodeUid, Result<()>)>>; + +/// Drive as the sync core sees it. See the module documentation. +#[async_trait] +pub(crate) trait DriveApi: Send + Sync { + async fn get_my_files_folder(&self) -> Result; + + async fn get_node(&self, uid: &NodeUid) -> Result>; + + async fn enumerate_nodes(&self, uids: &[NodeUid]) -> Result>; + + /// As [`DriveApi::enumerate_nodes`], but files carry no claimed size or + /// modification time; see [`ProtonDriveClient::enumerate_nodes_light`]. + async fn enumerate_nodes_light(&self, uids: &[NodeUid]) -> Result>; + + async fn enumerate_folder_children_node_uids( + &self, + folder_uid: &NodeUid, + ) -> Result>; + + async fn enumerate_events( + &self, + scope: &DriveEventScopeId, + cursor: Option<&DriveEventId>, + ) -> Result>; + + async fn invalidate_caches_for_event(&self, event: &DriveEvent) -> Result<()>; + + async fn create_folder( + &self, + parent_uid: &NodeUid, + name: &str, + last_modification_time: Option, + ) -> Result; + + async fn upload_file( + &self, + parent_uid: &NodeUid, + name: &str, + media_type: &str, + contents: &[u8], + ) -> Result; + + #[allow(clippy::too_many_arguments)] + async fn upload_file_from_dyn( + &self, + parent_uid: &NodeUid, + name: &str, + media_type: &str, + reader: &mut (dyn Read + Send), + intended_size: i64, + thumbnails: Vec, + last_modification_time: Option, + aead: bool, + ) -> Result; + + #[allow(clippy::too_many_arguments)] + async fn upload_file_replacing_draft_from_dyn( + &self, + parent_uid: &NodeUid, + name: &str, + media_type: &str, + reader: &mut (dyn Read + Send), + intended_size: i64, + thumbnails: Vec, + last_modification_time: Option, + aead: bool, + ) -> Result; + + async fn upload_new_revision_from_dyn( + &self, + file_uid: &NodeUid, + reader: &mut (dyn Read + Send), + intended_size: i64, + thumbnails: Vec, + last_modification_time: Option, + ) -> Result<()>; + + async fn open_revision(&self, uid: &NodeUid) -> Result>; + + async fn download_file_to_dyn( + &self, + uid: &NodeUid, + output: &mut (dyn Write + Send), + ) -> Result<()>; + + async fn download_revision_to_dyn( + &self, + file_uid: &NodeUid, + revision_id: &str, + writer: &mut (dyn Write + Send), + ) -> Result<()>; + + async fn rename_node( + &self, + uid: &NodeUid, + new_name: &str, + new_media_type: Option<&str>, + ) -> Result<()>; + + async fn move_node(&self, uid: &NodeUid, new_parent: &NodeUid) -> Result<()>; + + fn move_nodes_streaming<'a>( + &'a self, + items: Vec, + new_parent: NodeUid, + ) -> OutcomeStream<'a>; + + async fn trash_nodes(&self, uids: &[NodeUid]) -> Result; + + async fn restore_nodes(&self, uids: &[NodeUid]) -> Result; + + fn restore_nodes_streaming<'a>(&'a self, uids: &[NodeUid]) -> OutcomeStream<'a>; + + fn delete_nodes_streaming<'a>(&'a self, uids: &[NodeUid]) -> OutcomeStream<'a>; + + async fn enumerate_trash_node_uids(&self) -> Result>; +} + +/// The generic signatures of the client, over the object-safe ones above. +impl dyn DriveApi { + #[allow(clippy::too_many_arguments)] + pub(crate) async fn upload_file_from( + &self, + parent_uid: &NodeUid, + name: &str, + media_type: &str, + mut reader: R, + intended_size: i64, + thumbnails: Vec, + last_modification_time: Option, + aead: bool, + ) -> Result { + self.upload_file_from_dyn( + parent_uid, + name, + media_type, + &mut reader, + intended_size, + thumbnails, + last_modification_time, + aead, + ) + .await + } + + #[allow(clippy::too_many_arguments)] + pub(crate) async fn upload_file_replacing_draft_from( + &self, + parent_uid: &NodeUid, + name: &str, + media_type: &str, + mut reader: R, + intended_size: i64, + thumbnails: Vec, + last_modification_time: Option, + aead: bool, + ) -> Result { + self.upload_file_replacing_draft_from_dyn( + parent_uid, + name, + media_type, + &mut reader, + intended_size, + thumbnails, + last_modification_time, + aead, + ) + .await + } + + pub(crate) async fn upload_new_revision_from( + &self, + file_uid: &NodeUid, + mut reader: R, + intended_size: i64, + thumbnails: Vec, + last_modification_time: Option, + ) -> Result<()> { + self.upload_new_revision_from_dyn( + file_uid, + &mut reader, + intended_size, + thumbnails, + last_modification_time, + ) + .await + } + + pub(crate) async fn download_file_to( + &self, + uid: &NodeUid, + output: &mut W, + ) -> Result<()> { + self.download_file_to_dyn(uid, output).await + } + + pub(crate) async fn download_revision_to( + &self, + file_uid: &NodeUid, + revision_id: &str, + writer: &mut W, + ) -> Result<()> { + self.download_revision_to_dyn(file_uid, revision_id, writer) + .await + } +} + +/// An open revision that block reads are served from; see [`RevisionReader`]. +#[async_trait] +pub(crate) trait RevisionRead: Send + Sync { + /// Plaintext size of the revision. + fn size(&self) -> u64; + + /// Plaintext size of each block, in block order. + fn block_sizes(&self) -> &[u64]; + + async fn read_at(&self, offset: u64, length: u64) -> Result>; +} + +#[async_trait] +impl RevisionRead for RevisionReader { + fn size(&self) -> u64 { + RevisionReader::size(self) + } + + fn block_sizes(&self) -> &[u64] { + RevisionReader::block_sizes(self) + } + + async fn read_at(&self, offset: u64, length: u64) -> Result> { + RevisionReader::read_at(self, offset, length).await + } +} + +#[async_trait] +impl DriveApi for ProtonDriveClient { + async fn get_my_files_folder(&self) -> Result { + ProtonDriveClient::get_my_files_folder(self).await + } + + async fn get_node(&self, uid: &NodeUid) -> Result> { + ProtonDriveClient::get_node(self, uid).await + } + + async fn enumerate_nodes(&self, uids: &[NodeUid]) -> Result> { + ProtonDriveClient::enumerate_nodes(self, uids).await + } + + async fn enumerate_nodes_light(&self, uids: &[NodeUid]) -> Result> { + ProtonDriveClient::enumerate_nodes_light(self, uids).await + } + + async fn enumerate_folder_children_node_uids( + &self, + folder_uid: &NodeUid, + ) -> Result> { + ProtonDriveClient::enumerate_folder_children_node_uids(self, folder_uid).await + } + + async fn enumerate_events( + &self, + scope: &DriveEventScopeId, + cursor: Option<&DriveEventId>, + ) -> Result> { + ProtonDriveClient::enumerate_events(self, scope, cursor).await + } + + async fn invalidate_caches_for_event(&self, event: &DriveEvent) -> Result<()> { + ProtonDriveClient::invalidate_caches_for_event(self, event).await + } + + async fn create_folder( + &self, + parent_uid: &NodeUid, + name: &str, + last_modification_time: Option, + ) -> Result { + ProtonDriveClient::create_folder(self, parent_uid, name, last_modification_time).await + } + + async fn upload_file( + &self, + parent_uid: &NodeUid, + name: &str, + media_type: &str, + contents: &[u8], + ) -> Result { + ProtonDriveClient::upload_file(self, parent_uid, name, media_type, contents).await + } + + async fn upload_file_from_dyn( + &self, + parent_uid: &NodeUid, + name: &str, + media_type: &str, + reader: &mut (dyn Read + Send), + intended_size: i64, + thumbnails: Vec, + last_modification_time: Option, + aead: bool, + ) -> Result { + ProtonDriveClient::upload_file_from( + self, + parent_uid, + name, + media_type, + reader, + intended_size, + thumbnails, + last_modification_time, + aead, + ) + .await + } + + async fn upload_file_replacing_draft_from_dyn( + &self, + parent_uid: &NodeUid, + name: &str, + media_type: &str, + reader: &mut (dyn Read + Send), + intended_size: i64, + thumbnails: Vec, + last_modification_time: Option, + aead: bool, + ) -> Result { + ProtonDriveClient::upload_file_replacing_draft_from( + self, + parent_uid, + name, + media_type, + reader, + intended_size, + thumbnails, + last_modification_time, + aead, + ) + .await + } + + async fn upload_new_revision_from_dyn( + &self, + file_uid: &NodeUid, + reader: &mut (dyn Read + Send), + intended_size: i64, + thumbnails: Vec, + last_modification_time: Option, + ) -> Result<()> { + ProtonDriveClient::upload_new_revision_from( + self, + file_uid, + reader, + intended_size, + thumbnails, + last_modification_time, + ) + .await + } + + async fn open_revision(&self, uid: &NodeUid) -> Result> { + let reader = ProtonDriveClient::open_revision(self, uid).await?; + Ok(Arc::new(reader)) + } + + async fn download_file_to_dyn( + &self, + uid: &NodeUid, + output: &mut (dyn Write + Send), + ) -> Result<()> { + ProtonDriveClient::download_file_to(self, uid, &mut &mut *output).await + } + + async fn download_revision_to_dyn( + &self, + file_uid: &NodeUid, + revision_id: &str, + writer: &mut (dyn Write + Send), + ) -> Result<()> { + ProtonDriveClient::download_revision_to(self, file_uid, revision_id, &mut &mut *writer) + .await + } + + async fn rename_node( + &self, + uid: &NodeUid, + new_name: &str, + new_media_type: Option<&str>, + ) -> Result<()> { + ProtonDriveClient::rename_node(self, uid, new_name, new_media_type).await + } + + async fn move_node(&self, uid: &NodeUid, new_parent: &NodeUid) -> Result<()> { + ProtonDriveClient::move_node(self, uid, new_parent).await + } + + fn move_nodes_streaming<'a>( + &'a self, + items: Vec, + new_parent: NodeUid, + ) -> OutcomeStream<'a> { + ProtonDriveClient::move_nodes_streaming(self, items, new_parent).boxed() + } + + async fn trash_nodes(&self, uids: &[NodeUid]) -> Result { + ProtonDriveClient::trash_nodes(self, uids).await + } + + async fn restore_nodes(&self, uids: &[NodeUid]) -> Result { + ProtonDriveClient::restore_nodes(self, uids).await + } + + fn restore_nodes_streaming<'a>(&'a self, uids: &[NodeUid]) -> OutcomeStream<'a> { + ProtonDriveClient::restore_nodes_streaming(self, uids).boxed() + } + + fn delete_nodes_streaming<'a>(&'a self, uids: &[NodeUid]) -> OutcomeStream<'a> { + ProtonDriveClient::delete_nodes_streaming(self, uids).boxed() + } + + async fn enumerate_trash_node_uids(&self) -> Result> { + ProtonDriveClient::enumerate_trash_node_uids(self).await + } +} diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs index 820e68d..6cecdd4 100644 --- a/crates/pdfs-fuse/src/filesystem.rs +++ b/crates/pdfs-fuse/src/filesystem.rs @@ -1511,7 +1511,7 @@ impl ProtonFs { } // 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( + match self.core.block_on_bounded(self.core.drive.upload_file( &parent_uid, name, media_type_for(name), @@ -1617,7 +1617,7 @@ impl ProtonFs { } match self .core - .block_on_bounded(self.core.client.create_folder(&parent_uid, name, now)) + .block_on_bounded(self.core.drive.create_folder(&parent_uid, name, now)) { Ok(new_uid) => self.core.fetch_minted_node(&new_uid, name), Err(e) if self.core.lost_link(&e, "mkdir") => None, @@ -2143,7 +2143,7 @@ impl ProtonFs { } if let Err(e) = self .core - .block_on_bounded(self.core.client.trash_nodes(std::slice::from_ref(&uid))) + .block_on_bounded(self.core.drive.trash_nodes(std::slice::from_ref(&uid))) .and_then(batch::into_unit) { if self.core.lost_link(&e, "trash") { diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs index 100aa37..30c2f85 100644 --- a/crates/pdfs-fuse/src/lib.rs +++ b/crates/pdfs-fuse/src/lib.rs @@ -75,7 +75,7 @@ use proton_drive_rs::proton_sdk::error::ProtonError; use proton_drive_rs::proton_sdk::ids::{DriveEventId, LinkId, NodeUid, ShareId, VolumeId}; use proton_drive_rs::{ DriveEvent, DriveEventScopeId, MemberRole, Node, NodeKind, ProtonDriveClient, - ProtonPhotosClient, RevisionReader, SharedWithMeItem, ThumbnailType, + ProtonPhotosClient, SharedWithMeItem, ThumbnailType, }; mod albums; @@ -86,6 +86,8 @@ use control::run_control_socket; mod devices; mod diagnostics; mod drain; +mod drive; +use drive::{DriveApi, RevisionRead}; mod events; mod filesystem; mod link; @@ -484,6 +486,12 @@ fn relative_to<'a>(root: &str, path: &'a str) -> Option<&'a str> { #[derive(Clone)] struct Core { + /// Drive as the sync core uses it: listings, events, creates, uploads, + /// downloads, renames, moves and trashes. The real client in production, an + /// in-memory Drive in the simulation tests. See [`drive`]. + drive: Arc, + /// The concrete client, for what [`DriveApi`] does not cover: sharing, + /// Photos, devices, invitations and public links. client: ProtonDriveClient, rt: tokio::runtime::Handle, /// The daemon's My Files root. Forked on-demand sessions retain this value @@ -1618,14 +1626,14 @@ impl Core { /// name hash. See [`retry_out_of_date`]. pub(crate) fn rename_remote(&self, uid: &NodeUid, name: &str) -> Result<(), ProtonError> { self.change_remote(uid, || { - self.block_on_bounded(self.client.rename_node(uid, name, None)) + self.block_on_bounded(self.drive.rename_node(uid, name, None)) }) } /// Move `uid` into the folder `parent` on Drive, like [`Core::rename_remote`]. pub(crate) fn move_remote(&self, uid: &NodeUid, parent: &NodeUid) -> Result<(), ProtonError> { self.change_remote(uid, || { - self.block_on_bounded(self.client.move_node(uid, parent)) + self.block_on_bounded(self.drive.move_node(uid, parent)) }) } @@ -1657,7 +1665,7 @@ impl Core { is_trashed: false, is_shared: false, }; - if let Err(error) = self.block_on_bounded(self.client.invalidate_caches_for_event(&changed)) + if let Err(error) = self.block_on_bounded(self.drive.invalidate_caches_for_event(&changed)) { debug!(%uid, %error, "could not drop the SDK's cached node"); } @@ -2239,7 +2247,7 @@ impl Core { Vec::new() } else { self.rt - .block_on(self.client.enumerate_nodes_light(&accepted.uids)) + .block_on(self.drive.enumerate_nodes_light(&accepted.uids)) .map_err(|error| { error!(%error, "materializing shared roots failed"); Errno::EIO @@ -2445,7 +2453,7 @@ impl Core { refresh_foreign.then(|| self.shared_generation.load(Ordering::SeqCst)); let uids = match self .rt - .block_on(self.client.enumerate_folder_children_node_uids(&folder_uid)) + .block_on(self.drive.enumerate_folder_children_node_uids(&folder_uid)) { Ok(uids) => uids, Err(e) if self.lost_link(&e, "list folder") => { @@ -2465,7 +2473,7 @@ 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 = match self.rt.block_on(self.client.enumerate_nodes_light(&uids)) { + let mut nodes = match self.rt.block_on(self.drive.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); @@ -2732,7 +2740,7 @@ impl Core { let (mut key, mut uids, mut slot) = (key, uids, slot); loop { for chunk in uids.chunks(SIZE_UPGRADE_CHUNK) { - let result = self.rt.block_on(self.client.enumerate_nodes(chunk)); + let result = self.rt.block_on(self.drive.enumerate_nodes(chunk)); self.apply_size_upgrade(key, result); slot.chunk_done(); } @@ -2763,7 +2771,7 @@ impl Core { if !self.is_online() { return None; } - let result = self.block_on_bounded(self.client.enumerate_nodes(std::slice::from_ref(uid))); + let result = self.block_on_bounded(self.drive.enumerate_nodes(std::slice::from_ref(uid))); if let Err(e) = &result { self.lost_link(e, "settle a provisional size"); } @@ -3068,7 +3076,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.block_on_bounded(self.client.enumerate_nodes(std::slice::from_ref(uid))) { + match self.block_on_bounded(self.drive.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. @@ -3897,7 +3905,7 @@ impl Core { .begin(name, uid.to_string(), TransferDirection::Download, total); let mut out = CountingWriter::new(Vec::with_capacity(total as usize), &guard); self.rt - .block_on(self.client.download_file_to(uid, &mut out))?; + .block_on(self.drive.download_file_to(uid, &mut out))?; Ok(out.into_inner()) } @@ -3906,7 +3914,7 @@ impl Core { /// [`Core::download_file_tracked`], and preserves the same transfer /// accounting while allowing callers that only need an on-disk file to /// keep memory use independent of file size. - fn download_file_tracked_to( + fn download_file_tracked_to( &self, uid: &NodeUid, name: &str, @@ -3918,7 +3926,7 @@ impl Core { .begin(name, uid.to_string(), TransferDirection::Download, total); let mut out = CountingWriter::new(out, &guard); self.rt - .block_on(self.client.download_file_to(uid, &mut out))?; + .block_on(self.drive.download_file_to(uid, &mut out))?; Ok(out.into_inner()) } @@ -4447,7 +4455,7 @@ impl Core { Some(node) => node, None => self .rt - .block_on(self.client.get_node(&uid)) + .block_on(self.drive.get_node(&uid)) .map_err(|e| CoreError::from_api(&e, "get node"))? .ok_or_else(|| CoreError::not_found("node vanished"))?, }, @@ -4685,7 +4693,7 @@ impl Core { ) .map_err(|error| self.errno_error(error, "trash access"))?; self.rt - .block_on(self.client.trash_nodes(std::slice::from_ref(&uid))) + .block_on(self.drive.trash_nodes(std::slice::from_ref(&uid))) .and_then(batch::into_unit) .map_err(|e| CoreError::from_api(&e, "trash"))?; // Uploads queued for the node or anything below it would now only fail @@ -4738,7 +4746,7 @@ impl Core { } self.require_uid_writable(uid)?; if let Err(e) = self - .block_on_bounded(self.client.trash_nodes(std::slice::from_ref(uid))) + .block_on_bounded(self.drive.trash_nodes(std::slice::from_ref(uid))) .and_then(batch::into_unit) { if self.lost_link(&e, "trash a replaced node") { @@ -4811,7 +4819,7 @@ impl Core { } match self .rt - .block_on(self.client.restore_nodes(std::slice::from_ref(uid))) + .block_on(self.drive.restore_nodes(std::slice::from_ref(uid))) .and_then(batch::into_unit) { Ok(()) => { @@ -4893,7 +4901,7 @@ impl Core { /// end. Every batch is a usable listing. async fn refresh_trash(&self) -> CoreResult<()> { let started = Instant::now(); - let uids = within_trash_call("enumerate trash", self.client.enumerate_trash_node_uids()) + let uids = within_trash_call("enumerate trash", self.drive.enumerate_trash_node_uids()) .await? .map_err(|e| CoreError::from_api(&e, "enumerate trash"))?; info!( @@ -4920,7 +4928,7 @@ impl Core { let mut items: Vec = Vec::with_capacity(uids.len()); for chunk in uids.chunks(TRASH_MATERIALIZE_CHUNK) { let chunk_started = Instant::now(); - let nodes = within_trash_call("enumerate nodes", self.client.enumerate_nodes(chunk)) + let nodes = within_trash_call("enumerate nodes", self.drive.enumerate_nodes(chunk)) .await? .map_err(|e| CoreError::from_api(&e, "enumerate nodes"))?; items.extend(nodes.into_iter().map(|node| StoredTrash { @@ -5082,7 +5090,7 @@ impl Core { let all: Vec = waves.iter().flatten().cloned().collect(); let parents: Vec = self .rt - .block_on(self.client.enumerate_nodes(&all)) + .block_on(self.drive.enumerate_nodes(&all)) .map_err(|e| CoreError::from_api(&e, "enumerate nodes"))? .into_iter() .filter_map(|n| n.parent_uid) @@ -5098,7 +5106,7 @@ impl Core { for wave in &waves { self.rt .block_on(async { - let mut outcomes = std::pin::pin!(self.client.restore_nodes_streaming(wave)); + let mut outcomes = std::pin::pin!(self.drive.restore_nodes_streaming(wave)); while let Some(item) = outcomes.next().await { let (uid, outcome) = item?; match outcome { @@ -5146,7 +5154,7 @@ impl Core { let mut first_error: Option = None; self.rt .block_on(async { - let mut outcomes = std::pin::pin!(self.client.delete_nodes_streaming(&parsed)); + let mut outcomes = std::pin::pin!(self.drive.delete_nodes_streaming(&parsed)); while let Some(item) = outcomes.next().await { let (uid, outcome) = item?; match outcome { @@ -5178,7 +5186,7 @@ impl Core { fn empty_trash(&self) -> CoreResult { let uids = self .rt - .block_on(self.client.enumerate_trash_node_uids()) + .block_on(self.drive.enumerate_trash_node_uids()) .map_err(|e| CoreError::from_api(&e, "enumerate trash"))?; self.rt .block_on(self.client.empty_trash()) @@ -5220,7 +5228,7 @@ impl Core { let new_uid = self .rt .block_on( - self.client + self.drive .create_folder(&parent_uid, name, Some(now_secs())), ) .map_err(|e| CoreError::from_api(&e, "create folder"))?; @@ -8529,12 +8537,12 @@ mod tests { include_str!("relocate.rs"), ] { assert_eq!( - source.matches(concat!("client.rename_node", "(")).count(), + source.matches(concat!("drive.rename_node", "(")).count(), usize::from(source.contains("fn rename_remote(")), "call Core::rename_remote instead of the SDK rename directly" ); assert_eq!( - source.matches(concat!("client.move_node", "(")).count(), + source.matches(concat!("drive.move_node", "(")).count(), usize::from(source.contains("fn move_remote(")), "call Core::move_remote instead of the SDK move directly" ); diff --git a/crates/pdfs-fuse/src/link.rs b/crates/pdfs-fuse/src/link.rs index d08c1ba..66e94b9 100644 --- a/crates/pdfs-fuse/src/link.rs +++ b/crates/pdfs-fuse/src/link.rs @@ -250,7 +250,7 @@ impl Core { return false; } let probe = self.rt.block_on(async { - match tokio::time::timeout(PROBE_TIMEOUT, self.client.get_my_files_folder()).await { + match tokio::time::timeout(PROBE_TIMEOUT, self.drive.get_my_files_folder()).await { Ok(result) => result, Err(_) => Err(no_answer(PROBE_TIMEOUT)), } diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 3f02037..1de7b7f 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -364,6 +364,7 @@ pub fn mount( let paused_until = pause::load_paused_until(&db); let core = Core { + drive: Arc::new(client.clone()), client: client.clone(), rt: rt.clone(), maintenance: Arc::new(Mutex::new(Default::default())), @@ -625,7 +626,7 @@ pub fn mount( } } }; - rt.spawn(run_event_sync(client, scope, core.clone())); + rt.spawn(run_event_sync(core.drive.clone(), scope, core.clone())); // The photos volume has its own event stream; without it the gallery only // learns about a phone-side deletion when the timeline goes stale. rt.spawn(run_photos_event_sync(core.clone())); diff --git a/crates/pdfs-fuse/src/photos.rs b/crates/pdfs-fuse/src/photos.rs index a9486c3..519cb2d 100644 --- a/crates/pdfs-fuse/src/photos.rs +++ b/crates/pdfs-fuse/src/photos.rs @@ -1038,7 +1038,7 @@ impl Core { } let outcomes = self .rt - .block_on(self.client.trash_nodes(uids)) + .block_on(self.drive.trash_nodes(uids)) .map_err(|e| CoreError::from_api(&e, "trash"))?; let (trashed, failed) = pdfs_core::batch::split(outcomes); diff --git a/crates/pdfs-fuse/src/profile.rs b/crates/pdfs-fuse/src/profile.rs index 60f5068..4ebeffd 100644 --- a/crates/pdfs-fuse/src/profile.rs +++ b/crates/pdfs-fuse/src/profile.rs @@ -101,7 +101,7 @@ impl Core { match existing { Some(uid) => self .rt - .block_on(self.client.upload_new_revision_from( + .block_on(self.drive.upload_new_revision_from( &uid, Cursor::new(bytes), len, @@ -112,7 +112,7 @@ impl Core { .map_err(|e| CoreError::from_api(&e, "upload profile revision"))?, None => { self.rt - .block_on(self.client.upload_file_replacing_draft_from( + .block_on(self.drive.upload_file_replacing_draft_from( &dir_uid, PROFILE_FILE_NAME, "application/json", @@ -182,14 +182,14 @@ impl Core { ) -> CoreResult> { let uids = self .rt - .block_on(self.client.enumerate_folder_children_node_uids(parent_uid)) + .block_on(self.drive.enumerate_folder_children_node_uids(parent_uid)) .map_err(|e| CoreError::from_api(&e, "list device root"))?; if uids.is_empty() { return Ok(None); } let nodes = self .rt - .block_on(self.client.enumerate_nodes(&uids)) + .block_on(self.drive.enumerate_nodes(&uids)) .map_err(|e| CoreError::from_api(&e, "resolve device root children"))?; Ok(nodes .into_iter() @@ -210,7 +210,7 @@ impl Core { let uid = self .rt .block_on( - self.client + self.drive .create_folder(root_uid, PROFILE_DIR_NAME, Some(now_secs())), ) .map_err(|e| CoreError::from_api(&e, "create profile folder"))?; @@ -241,7 +241,7 @@ impl Core { }; let mut buf: Vec = Vec::new(); self.rt - .block_on(self.client.download_file_to(&uid, &mut buf)) + .block_on(self.drive.download_file_to(&uid, &mut buf)) .map_err(|e| CoreError::from_api(&e, "download profile"))?; Profile::parse(&buf).map(Some).map_err(CoreError::invalid) } @@ -277,13 +277,13 @@ impl Core { let uids = self .rt - .block_on(self.client.enumerate_folder_children_node_uids(&root_uid)) + .block_on(self.drive.enumerate_folder_children_node_uids(&root_uid)) .map_err(|e| CoreError::from_api(&e, "list device root"))?; let nodes = if uids.is_empty() { Vec::new() } else { self.rt - .block_on(self.client.enumerate_nodes(&uids)) + .block_on(self.drive.enumerate_nodes(&uids)) .map_err(|e| CoreError::from_api(&e, "resolve device root children"))? }; diff --git a/crates/pdfs-fuse/src/reads.rs b/crates/pdfs-fuse/src/reads.rs index 4d27b38..5172de4 100644 --- a/crates/pdfs-fuse/src/reads.rs +++ b/crates/pdfs-fuse/src/reads.rs @@ -70,7 +70,7 @@ const MAX_OPEN_READERS: usize = 64; /// `(mtime, size)` — the same validity pair the content cache uses (a new /// revision bumps mtime). On a mismatch the reader is dropped and reopened. pub(super) struct CachedReader { - reader: Arc, + reader: Arc, mtime: i64, size: u64, /// For LRU eviction. @@ -101,7 +101,7 @@ pub(super) struct PendingOpen { /// must not join this open — it would get a reader for the wrong bytes. mtime: i64, size: u64, - rx: tokio::sync::watch::Receiver, Errno>>>, + rx: tokio::sync::watch::Receiver, Errno>>>, } /// Distinguishes `PendingOpen`s. Only uniqueness matters, so a plain counter @@ -313,16 +313,16 @@ impl Core { uid: &NodeUid, mtime: i64, fsize: u64, - ) -> Result, Errno> { + ) -> Result, Errno> { loop { // Decide under the lock, act outside it — network I/O must never run // under a std `Mutex`, and an await must never hold one. enum Act { - Hit(Arc), - Join(tokio::sync::watch::Receiver, Errno>>>), + Hit(Arc), + Join(tokio::sync::watch::Receiver, Errno>>>), Lead( u64, - tokio::sync::watch::Sender, Errno>>>, + tokio::sync::watch::Sender, Errno>>>, ), } let act = { @@ -376,8 +376,8 @@ impl Core { } Act::Lead(id, tx) => { debug!(%uid, mtime, fsize, "opening revision"); - let result = match self.client.open_revision(uid).await { - Ok(reader) => Ok(Arc::new(reader)), + let result = match self.drive.open_revision(uid).await { + Ok(reader) => Ok(reader), Err(e) => { self.lost_link(&e, "open a revision"); warn!(%uid, error = %e, "open_revision failed"); @@ -568,7 +568,7 @@ impl Core { /// leader's claim is gone by then, so this cannot recurse. async fn fetch_block( &self, - reader: &Arc, + reader: &Arc, uid: &NodeUid, mtime: i64, fsize: u64, @@ -634,7 +634,7 @@ impl Core { /// download, or the read fails. See [`Core::repair_block`] (bugs.md B84). async fn read_block( &self, - reader: &Arc, + reader: &Arc, uid: &NodeUid, mtime: i64, fsize: u64, @@ -778,7 +778,7 @@ impl Core { error!(%uid, %error, "creating a scratch file for a repair download failed"); Errno::EIO })?; - let outcome = self.client.download_file_to(uid, &mut file).await; + let outcome = self.drive.download_file_to(uid, &mut file).await; let stored = (|| { outcome.map_err(|error| { warn!(%uid, %error, "repair download failed"); diff --git a/crates/pdfs-fuse/src/redate.rs b/crates/pdfs-fuse/src/redate.rs index 89f9488..7f86890 100644 --- a/crates/pdfs-fuse/src/redate.rs +++ b/crates/pdfs-fuse/src/redate.rs @@ -248,10 +248,7 @@ impl Core { let outcomes = self .rt - .block_on( - self.client - .trash_nodes(std::slice::from_ref(&candidate.uid)), - ) + .block_on(self.drive.trash_nodes(std::slice::from_ref(&candidate.uid))) .map_err(|e| CoreError::from_api(&e, "trash the mis-dated original"))?; if let Some((_, Err(e))) = outcomes.into_iter().next() { return Err(CoreError::from_api(&e, "trash the mis-dated original")); diff --git a/crates/pdfs-fuse/src/revisions.rs b/crates/pdfs-fuse/src/revisions.rs index 240b170..2aa1d83 100644 --- a/crates/pdfs-fuse/src/revisions.rs +++ b/crates/pdfs-fuse/src/revisions.rs @@ -232,7 +232,7 @@ impl Core { let mut out = std::io::BufWriter::new(file); let written = match self .rt - .block_on(self.client.download_revision_to(uid, revision_id, &mut out)) + .block_on(self.drive.download_revision_to(uid, revision_id, &mut out)) .map_err(|e| CoreError::from_api(&e, "download revision")) .and_then(|()| { use std::io::Write as _; diff --git a/crates/pdfs-fuse/src/sharing.rs b/crates/pdfs-fuse/src/sharing.rs index 65b7df2..c22ff0b 100644 --- a/crates/pdfs-fuse/src/sharing.rs +++ b/crates/pdfs-fuse/src/sharing.rs @@ -463,7 +463,7 @@ impl Core { } let nodes = self .rt - .block_on(self.client.enumerate_nodes(&uids)) + .block_on(self.drive.enumerate_nodes(&uids)) .map_err(|e| CoreError::from_api(&e, "enumerate nodes"))?; Ok(nodes .into_iter() @@ -481,14 +481,14 @@ impl Core { parse_uid(uid).ok_or_else(|| CoreError::invalid(format!("invalid uid: {uid}")))?; let child_uids = self .rt - .block_on(self.client.enumerate_folder_children_node_uids(&uid)) + .block_on(self.drive.enumerate_folder_children_node_uids(&uid)) .map_err(|e| CoreError::from_api(&e, "enumerate shared children"))?; if child_uids.is_empty() { return Ok(Vec::new()); } let nodes = self .rt - .block_on(self.client.enumerate_nodes(&child_uids)) + .block_on(self.drive.enumerate_nodes(&child_uids)) .map_err(|e| CoreError::from_api(&e, "enumerate nodes"))?; Ok(nodes .into_iter() @@ -509,7 +509,7 @@ impl Core { parse_uid(uid).ok_or_else(|| CoreError::invalid(format!("invalid uid: {uid}")))?; let node = self .rt - .block_on(self.client.get_node(&uid)) + .block_on(self.drive.get_node(&uid)) .map_err(|e| CoreError::from_api(&e, "get shared node"))? .ok_or_else(|| CoreError::not_found("shared node not found"))?; if !node.is_file() { @@ -669,7 +669,7 @@ impl Core { } let nodes = self .rt - .block_on(self.client.enumerate_nodes(&uids)) + .block_on(self.drive.enumerate_nodes(&uids)) .map_err(|e| CoreError::from_api(&e, "enumerate nodes"))?; let mut items = Vec::with_capacity(nodes.len()); for n in nodes { diff --git a/crates/pdfs-fuse/src/sweep.rs b/crates/pdfs-fuse/src/sweep.rs index 5e1422b..7b1e612 100644 --- a/crates/pdfs-fuse/src/sweep.rs +++ b/crates/pdfs-fuse/src/sweep.rs @@ -165,7 +165,7 @@ impl Core { } match self .rt - .block_on(self.client.trash_nodes(std::slice::from_ref(uid))) + .block_on(self.drive.trash_nodes(std::slice::from_ref(uid))) .and_then(batch::into_unit) { Ok(()) => {} diff --git a/crates/pdfs-fuse/src/sync.rs b/crates/pdfs-fuse/src/sync.rs index 52eef56..8425a51 100644 --- a/crates/pdfs-fuse/src/sync.rs +++ b/crates/pdfs-fuse/src/sync.rs @@ -370,7 +370,7 @@ impl Core { HashMap::new() } else { self.rt - .block_on(self.client.enumerate_nodes_light(&check)) + .block_on(self.drive.enumerate_nodes_light(&check)) .map_err(|e| format!("resolve changed nodes: {e}"))? .into_iter() .filter(|n| !n.trashed) @@ -525,7 +525,7 @@ impl Core { }; if let Err(e) = self .rt - .block_on(self.client.trash_nodes(&[uid])) + .block_on(self.drive.trash_nodes(&[uid])) .and_then(batch::into_unit) { warn!(rel, error = %e, "sync: trash remote failed"); @@ -776,7 +776,7 @@ impl Core { let remote = r.expect("remote deletion requires a remote file"); if let Err(e) = self .rt - .block_on(self.client.trash_nodes(std::slice::from_ref(&remote.uid))) + .block_on(self.drive.trash_nodes(std::slice::from_ref(&remote.uid))) .and_then(batch::into_unit) { warn!(rel, error = %e, "sync: trash remote failed"); @@ -839,7 +839,7 @@ impl Core { for (rel, uid) in delete_remote_dirs { if let Err(e) = self .rt - .block_on(self.client.trash_nodes(&[uid])) + .block_on(self.drive.trash_nodes(&[uid])) .and_then(batch::into_unit) { warn!(rel, error = %e, "sync: trash remote folder failed"); @@ -894,14 +894,14 @@ impl Core { ) -> Result<(), String> { let uids = self .rt - .block_on(self.client.enumerate_folder_children_node_uids(folder)) + .block_on(self.drive.enumerate_folder_children_node_uids(folder)) .map_err(|e| format!("enumerate {folder}: {e}"))?; if uids.is_empty() { return Ok(()); } let nodes = self .rt - .block_on(self.client.enumerate_nodes_light(&uids)) + .block_on(self.drive.enumerate_nodes_light(&uids)) .map_err(|e| format!("resolve nodes: {e}"))?; for node in nodes { if node.trashed || !node.is_folder() { @@ -946,14 +946,14 @@ impl Core { } let uids = self .rt - .block_on(self.client.enumerate_folder_children_node_uids(folder)) + .block_on(self.drive.enumerate_folder_children_node_uids(folder)) .map_err(|e| format!("enumerate {folder}: {e}"))?; if uids.is_empty() { return Ok(()); } let nodes = self .rt - .block_on(self.client.enumerate_nodes_light(&uids)) + .block_on(self.drive.enumerate_nodes_light(&uids)) .map_err(|e| format!("resolve nodes: {e}"))?; // Files whose recorded mtime no longer matches: their size has to be @@ -971,7 +971,7 @@ impl Core { HashMap::new() } else { self.rt - .block_on(self.client.enumerate_nodes(&stale)) + .block_on(self.drive.enumerate_nodes(&stale)) .map_err(|e| format!("resolve nodes: {e}"))? .iter() .map(|n| (n.uid.to_string(), remote_file_sig(n).1)) @@ -1139,7 +1139,7 @@ impl Core { match op { Pending::CreateDir { rel, parent } => { let uid = self - .client + .drive .create_folder(parent, base_name(rel), Some(now_secs())) .await .map_err(|e| format!("create remote folder {rel}: {e}"))?; @@ -1240,7 +1240,7 @@ impl Core { .begin(name, "", TransferDirection::Upload, meta.len()), ); let uid = self - .client + .drive // Mirror push: the local file is authoritative. If a prior attempt was // interrupted mid-upload it left an unsealed draft of this name; recover // it even across a daemon restart (which rotates our client uid), so the @@ -1294,7 +1294,7 @@ impl Core { meta.len(), ), ); - self.client + self.drive .upload_new_revision_from(uid, reader, meta.len() as i64, thumbnails, Some(mtime)) .await .map_err(|e| format!("upload revision {rel}: {e}"))?; @@ -1359,7 +1359,7 @@ impl Core { size.max(0) as u64, ); let mut out = CountingWriter::new(out, &guard); - self.client + self.drive .download_file_to(uid, &mut out) .await .map_err(|e| { @@ -1424,7 +1424,7 @@ impl Core { let (lmtime, lsize) = (local.mtime, local.size); // Re-fetch the node so the stored remote signature is exactly what a walk // will report next time. - let (rmtime, rsize) = match self.client.enumerate_nodes(std::slice::from_ref(uid)).await { + let (rmtime, rsize) = match self.drive.enumerate_nodes(std::slice::from_ref(uid)).await { Ok(nodes) => nodes .first() .map(remote_file_sig) diff --git a/crates/pdfs-fuse/src/upload.rs b/crates/pdfs-fuse/src/upload.rs index 1b8afb7..d9b03ca 100644 --- a/crates/pdfs-fuse/src/upload.rs +++ b/crates/pdfs-fuse/src/upload.rs @@ -73,7 +73,7 @@ async fn run_uploads( .upload_thumbnails_async(t.path.clone(), t.name.clone()) .await; match core - .client + .drive .upload_file_from( &t.parent_uid, &t.name, @@ -238,10 +238,7 @@ impl Core { .map_err(|error| self.errno_error(error, "upload destination"))?; let new_uid = self .rt - .block_on( - self.client - .create_folder(parent_uid, name, Some(now_secs())), - ) + .block_on(self.drive.create_folder(parent_uid, name, Some(now_secs()))) .map_err(|e| CoreError::from_api(&e, &format!("create folder {name}")))?; let node = self .fetch_minted_node(&new_uid, name)