refactor(fuse): reach Drive through a DriveApi trait

`Core` held a concrete `ProtonDriveClient`, so no test could make Drive
answer late, answer stale or not answer. This is the first step of
phase 0 in docs/MILESTONE-3.0.0.md.

`DriveApi` (`crates/pdfs-fuse/src/drive.rs`) covers the calls the sync
core makes: listings, events, creates, uploads, downloads, renames,
moves, trashes and restores. `ProtonDriveClient` implements it by
forwarding. `Core::drive` holds it, and every call of those methods now
goes through it. Sharing, Photos, devices, invitations and public links
stay on `Core::client`.

The trait takes readers and writers as trait objects, and
`impl dyn DriveApi` puts the client's generic signatures back, so call
sites read the same. Block reads go through `RevisionRead` instead of
the SDK's `RevisionReader`, which only the SDK can construct.

No behaviour change.
This commit is contained in:
nvrl committed 2026-10-03 04:37:37 +02:00
1 parent 8d92c69279
commit c73d5c9436
19 files changed
+560 -100

No files matched your search

Generated
+1
View File
@@ -3515,6 +3515,7 @@ dependencies = [
name = "pdfs-fuse"
version = "2.8.2"
dependencies = [
"async-trait",
"base64 0.23.1",
"fuser",
"futures",
+1
View File
@@ -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
+1 -1
View File
@@ -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<dyn DriveApi>,
scope: DriveEventScopeId,
core: Core,
) {
+6 -9
View File
@@ -363,14 +363,14 @@ impl Core {
) -> CoreResult<Option<NodeUid>> {
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,
+9 -9
View File
@@ -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<Option<NodeUid>, Box<dyn std::error::Error>> {
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<NodeUid, Box<dyn std::error::Error>> {
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,
+458
View File
@@ -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<Node>;
async fn get_node(&self, uid: &NodeUid) -> Result<Option<Node>>;
async fn enumerate_nodes(&self, uids: &[NodeUid]) -> Result<Vec<Node>>;
/// 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<Vec<Node>>;
async fn enumerate_folder_children_node_uids(
&self,
folder_uid: &NodeUid,
) -> Result<Vec<NodeUid>>;
async fn enumerate_events(
&self,
scope: &DriveEventScopeId,
cursor: Option<&DriveEventId>,
) -> Result<Vec<DriveEvent>>;
async fn invalidate_caches_for_event(&self, event: &DriveEvent) -> Result<()>;
async fn create_folder(
&self,
parent_uid: &NodeUid,
name: &str,
last_modification_time: Option<i64>,
) -> Result<NodeUid>;
async fn upload_file(
&self,
parent_uid: &NodeUid,
name: &str,
media_type: &str,
contents: &[u8],
) -> Result<NodeUid>;
#[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<Thumbnail>,
last_modification_time: Option<i64>,
aead: bool,
) -> Result<NodeUid>;
#[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<Thumbnail>,
last_modification_time: Option<i64>,
aead: bool,
) -> Result<NodeUid>;
async fn upload_new_revision_from_dyn(
&self,
file_uid: &NodeUid,
reader: &mut (dyn Read + Send),
intended_size: i64,
thumbnails: Vec<Thumbnail>,
last_modification_time: Option<i64>,
) -> Result<()>;
async fn open_revision(&self, uid: &NodeUid) -> Result<Arc<dyn RevisionRead>>;
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<NodeMoveItem>,
new_parent: NodeUid,
) -> OutcomeStream<'a>;
async fn trash_nodes(&self, uids: &[NodeUid]) -> Result<NodeOutcomes>;
async fn restore_nodes(&self, uids: &[NodeUid]) -> Result<NodeOutcomes>;
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<Vec<NodeUid>>;
}
/// 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<R: Read + Send>(
&self,
parent_uid: &NodeUid,
name: &str,
media_type: &str,
mut reader: R,
intended_size: i64,
thumbnails: Vec<Thumbnail>,
last_modification_time: Option<i64>,
aead: bool,
) -> Result<NodeUid> {
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<R: Read + Send>(
&self,
parent_uid: &NodeUid,
name: &str,
media_type: &str,
mut reader: R,
intended_size: i64,
thumbnails: Vec<Thumbnail>,
last_modification_time: Option<i64>,
aead: bool,
) -> Result<NodeUid> {
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<R: Read + Send>(
&self,
file_uid: &NodeUid,
mut reader: R,
intended_size: i64,
thumbnails: Vec<Thumbnail>,
last_modification_time: Option<i64>,
) -> 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<W: Write + Send>(
&self,
uid: &NodeUid,
output: &mut W,
) -> Result<()> {
self.download_file_to_dyn(uid, output).await
}
pub(crate) async fn download_revision_to<W: Write + Send>(
&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<Vec<u8>>;
}
#[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<Vec<u8>> {
RevisionReader::read_at(self, offset, length).await
}
}
#[async_trait]
impl DriveApi for ProtonDriveClient {
async fn get_my_files_folder(&self) -> Result<Node> {
ProtonDriveClient::get_my_files_folder(self).await
}
async fn get_node(&self, uid: &NodeUid) -> Result<Option<Node>> {
ProtonDriveClient::get_node(self, uid).await
}
async fn enumerate_nodes(&self, uids: &[NodeUid]) -> Result<Vec<Node>> {
ProtonDriveClient::enumerate_nodes(self, uids).await
}
async fn enumerate_nodes_light(&self, uids: &[NodeUid]) -> Result<Vec<Node>> {
ProtonDriveClient::enumerate_nodes_light(self, uids).await
}
async fn enumerate_folder_children_node_uids(
&self,
folder_uid: &NodeUid,
) -> Result<Vec<NodeUid>> {
ProtonDriveClient::enumerate_folder_children_node_uids(self, folder_uid).await
}
async fn enumerate_events(
&self,
scope: &DriveEventScopeId,
cursor: Option<&DriveEventId>,
) -> Result<Vec<DriveEvent>> {
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<i64>,
) -> Result<NodeUid> {
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<NodeUid> {
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<Thumbnail>,
last_modification_time: Option<i64>,
aead: bool,
) -> Result<NodeUid> {
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<Thumbnail>,
last_modification_time: Option<i64>,
aead: bool,
) -> Result<NodeUid> {
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<Thumbnail>,
last_modification_time: Option<i64>,
) -> 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<Arc<dyn RevisionRead>> {
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<NodeMoveItem>,
new_parent: NodeUid,
) -> OutcomeStream<'a> {
ProtonDriveClient::move_nodes_streaming(self, items, new_parent).boxed()
}
async fn trash_nodes(&self, uids: &[NodeUid]) -> Result<NodeOutcomes> {
ProtonDriveClient::trash_nodes(self, uids).await
}
async fn restore_nodes(&self, uids: &[NodeUid]) -> Result<NodeOutcomes> {
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<Vec<NodeUid>> {
ProtonDriveClient::enumerate_trash_node_uids(self).await
}
}
+3 -3
View File
@@ -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") {
+34 -26
View File
@@ -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<dyn DriveApi>,
/// 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<Option<Node>, 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<W: Write>(
fn download_file_tracked_to<W: Write + Send>(
&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<StoredTrash> = 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<NodeUid> = waves.iter().flatten().cloned().collect();
let parents: Vec<NodeUid> = 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<ProtonError> = 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<usize> {
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"
);
+1 -1
View File
@@ -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)),
}
+2 -1
View File
@@ -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()));
+1 -1
View File
@@ -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);
+8 -8
View File
@@ -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<Option<NodeUid>> {
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<u8> = 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"))?
};
+11 -11
View File
@@ -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<RevisionReader>,
reader: Arc<dyn RevisionRead>,
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<Option<Result<Arc<RevisionReader>, Errno>>>,
rx: tokio::sync::watch::Receiver<Option<Result<Arc<dyn RevisionRead>, Errno>>>,
}
/// Distinguishes `PendingOpen`s. Only uniqueness matters, so a plain counter
@@ -313,16 +313,16 @@ impl Core {
uid: &NodeUid,
mtime: i64,
fsize: u64,
) -> Result<Arc<RevisionReader>, Errno> {
) -> Result<Arc<dyn RevisionRead>, 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<RevisionReader>),
Join(tokio::sync::watch::Receiver<Option<Result<Arc<RevisionReader>, Errno>>>),
Hit(Arc<dyn RevisionRead>),
Join(tokio::sync::watch::Receiver<Option<Result<Arc<dyn RevisionRead>, Errno>>>),
Lead(
u64,
tokio::sync::watch::Sender<Option<Result<Arc<RevisionReader>, Errno>>>,
tokio::sync::watch::Sender<Option<Result<Arc<dyn RevisionRead>, 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<RevisionReader>,
reader: &Arc<dyn RevisionRead>,
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<RevisionReader>,
reader: &Arc<dyn RevisionRead>,
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");
+1 -4
View File
@@ -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"));
+1 -1
View File
@@ -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 _;
+5 -5
View File
@@ -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 {
+1 -1
View File
@@ -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(()) => {}
+14 -14
View File
@@ -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)
+2 -5
View File
@@ -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)