diff --git a/Cargo.lock b/Cargo.lock index 7217ad0..a4915c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2030,7 +2030,7 @@ dependencies = [ [[package]] name = "proton-drive-rs" -version = "0.1.12" +version = "0.2.0" dependencies = [ "async-trait", "base64", @@ -2055,7 +2055,7 @@ dependencies = [ [[package]] name = "proton-sdk" -version = "0.1.12" +version = "0.2.0" dependencies = [ "aes-gcm 0.11.0", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 14969e1..f4f0182 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "2" members = ["crates/proton-sdk", "crates/proton-drive-rs"] [workspace.package] -version = "0.1.12" +version = "0.2.0" edition = "2024" license = "MIT" repository = "https://github.com/narrrl/proton-sdk-rs" diff --git a/UPSTREAM_SYNC.md b/UPSTREAM_SYNC.md index 6d42977..aa414d4 100644 --- a/UPSTREAM_SYNC.md +++ b/UPSTREAM_SYNC.md @@ -7,8 +7,8 @@ gitignored, so this file is the only durable record of what we last reviewed. - **Upstream**: https://github.com/ProtonDriveApps/sdk - **Reconciled subtree**: `client/cs/src` (was `client/cs/sdk/src` before the mid-2026 projects reorg; the script watches both) -- **Pinned**: `1b6ff83566d73b331e6842f683f1c9afe2d5b4ee` -- **Date**: 2026-07-22 +- **Pinned**: `e0ad37e30ab16dd78899d1f4475fd6760b4c5c43` +- **Date**: 2026-07-24 > **Note**: upstream force-pushes — the SHAs in rows before 2026-07-14 no longer > exist in its history. Triage by commit subject/date when a pin goes stale. @@ -29,4 +29,5 @@ gitignored, so this file is the only durable record of what we last reviewed. | 2026-06-26 | `fb4173ab` → `36430318` | `36430318` | ported: enumeration returns `NodeUid`s (`enumerate_folder_children_node_uids` / `enumerate_trash_node_uids`), caller materializes via `enumerate_nodes`. noise dropped: BOM/deps/kt-enum-order/cs-account-refactor | | 2026-07-10 | `36430318` → `f2496161` | `9a1c39b3` | ported: validation check in `Thumbnail::new` returning `Result` instead of panicking. noise dropped: projects reorg / C# Account client move to incubating. deferred: device support (`32a8eed0`). | | 2026-07-14 | `f2496161` → `2219a42f` | `d15fa9b9`, `bead8f65`, `f6b12911`, `c0940f52`, `b88b8041`, `2eb69c46` | upstream force-pushed + moved the cs subtree, so the old pin was dangling and the script silently reported "up to date" (both fixed). ported: **devices** (`devices.rs` + `enumerate_devices` / `create_device` / `rename_device` / `delete_device`; volume-free `DeviceUid`), **node sharing flags** (`Node::is_shared` / `is_shared_publicly` from the link `Sharing` block), **revision state** (`NodeKind::File::active_revision_state`), **shared-with-me** (`enumerate_shared_with_me_node_uids`, `GET v2/sharedwithme`), **leave shared node** (`leave_shared_node`, `DELETE v2/shares/{sid}/members/{mid}`). no-op: `15fa845e` (non-JSON error bodies — `http.rs` already falls back to a status-only error when the envelope won't parse). diverged deliberately: `9a7f1d02` + `b4dadf02` removed the C# entity cache and its tags; ours stays. cosmetic, not ported: `30ab32c1` / `388d3900` (C# field renames), `1a781ade` (buffer lengths). | +| 2026-07-24 | `1b6ff835` → `e0ad37e3` | `d2ffde20`, `724b6771` | both features were already ported from the TypeScript SDK, so this closed the remaining parity deltas. ported: **shared-by-me on the photos volume** (`SharingOperations.EnumerateSharedNodeUidsAsync` now takes a volume resolver — `page_shared_by_me(volume_id)` + `ProtonPhotosClient::enumerate_shared_node_uids`, empty when the account has no photos volume) and `SharedByMeLinkDto.ContextShareID` (kept `Option` — recent upstream addition, older backends omit it); **per-node move outcomes** (`move_nodes` now returns `Vec<(NodeUid, Result<()>)>` in input order, mirroring the breaking `MoveNodesAsync` result map; per-link envelope codes via the new `aggregate_outcomes`, which `check_aggregate` now sits on). cosmetic, not ported: `AnchorID` retyped `string` → `LinkId` (ours is `String`). diverged deliberately: we keep the batch endpoint `PUT volumes/{vid}/links/move-multiple` — upstream still loops per node behind a `// FIXME: … use the batch move endpoint`; and the drive-side listing keeps the `_by_me` suffix (upstream renamed both to `EnumerateSharedNodeUids`) because `proton-drive-linux` already consumes that name. | | 2026-07-22 | `2219a42f` → `1b6ff835` | `ded1b716`, `071f976f` | ported: **atomic small uploads** for buffered new files and revisions (`DriveSmallFileUpload` is an explicit opt-in because this crate has no remote feature-flag provider), and **photo duplicate detection** using name/content HMAC matching. already covered/no-op: `ee0b3fba` (Rust rename accepts `Option<&str>`), `193c26f1` (Rust in-memory cache already has bounded LRU eviction), `29d359e0` (digest representation refactor retains hex JSON), `5c370289` (C# disposal lifetime only), `95752055` (Photos cache construction only). deliberately not ported: `2bf49aae` / `2590d227` cache storage-format changes; Rust exposes a distinct string repository contract and ships no persistent backend to migrate/reset. | diff --git a/crates/proton-drive-rs/Cargo.toml b/crates/proton-drive-rs/Cargo.toml index ca8a131..6ba445b 100644 --- a/crates/proton-drive-rs/Cargo.toml +++ b/crates/proton-drive-rs/Cargo.toml @@ -14,7 +14,7 @@ readme = "../../README.md" documentation = "https://docs.rs/proton-drive-rs" [dependencies] -proton-sdk = { path = "../proton-sdk", version = "0.1.12" } +proton-sdk = { path = "../proton-sdk", version = "0.2.0" } tokio.workspace = true reqwest.workspace = true serde.workspace = true diff --git a/crates/proton-drive-rs/src/client.rs b/crates/proton-drive-rs/src/client.rs index 49d4b42..14020e9 100644 --- a/crates/proton-drive-rs/src/client.rs +++ b/crates/proton-drive-rs/src/client.rs @@ -8,10 +8,11 @@ use std::sync::atomic::{AtomicBool, Ordering}; use base64::Engine; use base64::engine::general_purpose::STANDARD as BASE64; +use bytes::Bytes; use futures::stream::{self, StreamExt, TryStreamExt}; use lru::LruCache; use sha2::{Digest, Sha256}; -use tokio::sync::{Mutex, Semaphore}; +use tokio::sync::{Mutex, OwnedSemaphorePermit, Semaphore, mpsc}; use proton_sdk::account::{AccountClient, KeySalt}; use proton_sdk::api::ResponseCode; @@ -23,7 +24,7 @@ use proton_sdk::crypto::{ derive_key_passphrase, encrypt_external_invitation, encrypt_invitation, generate_node_hash_key, generate_node_key, generate_node_key_aead, generate_verifier, verify_detached, }; -use proton_sdk::error::{ProtonError, Result}; +use proton_sdk::error::{ProtonApiError, ProtonError, Result}; use proton_sdk::http::ApiHttpClient; use proton_sdk::ids::{ AddressId, AddressKeyId, DeviceUid, DriveEventId, LinkId, NodeUid, ShareId, ShareMembershipId, @@ -43,30 +44,30 @@ use crate::crypto::{ use crate::devices::{Device, DeviceMetadata, DeviceType}; use crate::dtos::{ AcceptInvitationRequest, AggregateLinksResponse, BlockCreationRequest, BlockDto, - BlockUploadPreparationRequest, BlockUploadPreparationResponse, BlockVerificationInputResponse, - BlockVerifier, BookmarkShareUrlDto, BookmarksResponse, CommonExtendedAttributes, - CreateBookmarkRequest, CreatePublicLinkRequest, CreatePublicLinkResponse, CreateShareRequest, - CreateShareResponse, DeviceCreationDeviceDto, DeviceCreationLinkDto, DeviceCreationRequest, - DeviceCreationResponse, DeviceCreationShareDto, DeviceListResponse, DeviceUpdateRequest, - DeviceUpdateShareDto, ExtendedAttributes, ExternalInvitationDto, ExternalInvitationResponseDto, - ExternalInvitationsResponse, FileContentDigests, FileCreationRequest, FileCreationResponse, - FindPhotoDuplicatesRequest, FindPhotoDuplicatesResponse, FolderChildrenResponse, - FolderCreationRequest, FolderCreationResponse, InvitationDetailsResponse, - InvitationsListResponse, InviteEmailDetailsDto, InviteExternalUserRequest, - InviteExternalUserResponse, InviteProtonUserInvitationDto, InviteProtonUserRequest, - InviteProtonUserResponse, LatestVolumeEventResponse, LinkDetailsDto, LinkDetailsRequest, - LinkDetailsResponse, LinkDto, LinkType, ModulusResponse, MoveLinkRequest, - MoveMultipleLinksItem, MoveMultipleLinksRequest, MultipleLinksRequest, MyFilesShareResponse, - NodeNameAvailabilityRequest, NodeNameAvailabilityResponse, PhotosAttributesDto, - RenameLinkRequest, RevisionConflict, RevisionCreationRequest, RevisionCreationResponse, - RevisionDto, RevisionListItemDto, RevisionListResponse, RevisionMetadataResponse, - RevisionResponse, RevisionUpdateRequest, ShareInvitationDto, ShareInvitationsResponse, - ShareMembersResponse, ShareResponse, ShareTargetType, ShareUrlDto, ShareUrlsResponse, - SharedByMeResponse, SharedWithMeResponse, SmallFileUploadMetadataRequest, - SmallRevisionUploadMetadataRequest, SmallUploadResponse, ThumbnailBlockListRequest, - ThumbnailBlockListResponse, ThumbnailCreationRequest, ThumbnailDto, TimelinePhotoListResponse, - UpdatePermissionsRequest, VolumeCreationRequest, VolumeEventDto, VolumeEventListResponse, - VolumeTrashResponse, + BlockUploadPreparationRequest, BlockUploadPreparationResponse, BlockUploadTarget, + BlockVerificationInputResponse, BlockVerifier, BookmarkShareUrlDto, BookmarksResponse, + CommonExtendedAttributes, CreateBookmarkRequest, CreatePublicLinkRequest, + CreatePublicLinkResponse, CreateShareRequest, CreateShareResponse, DeviceCreationDeviceDto, + DeviceCreationLinkDto, DeviceCreationRequest, DeviceCreationResponse, DeviceCreationShareDto, + DeviceListResponse, DeviceUpdateRequest, DeviceUpdateShareDto, ExtendedAttributes, + ExternalInvitationDto, ExternalInvitationResponseDto, ExternalInvitationsResponse, + FileContentDigests, FileCreationRequest, FileCreationResponse, FindPhotoDuplicatesRequest, + FindPhotoDuplicatesResponse, FolderChildrenResponse, FolderCreationRequest, + FolderCreationResponse, InvitationDetailsResponse, InvitationsListResponse, + InviteEmailDetailsDto, InviteExternalUserRequest, InviteExternalUserResponse, + InviteProtonUserInvitationDto, InviteProtonUserRequest, InviteProtonUserResponse, + LatestVolumeEventResponse, LinkDetailsDto, LinkDetailsRequest, LinkDetailsResponse, LinkDto, + LinkType, ModulusResponse, MoveLinkRequest, MoveMultipleLinksItem, MoveMultipleLinksRequest, + MultipleLinksRequest, MyFilesShareResponse, NodeNameAvailabilityRequest, + NodeNameAvailabilityResponse, PhotosAttributesDto, RenameLinkRequest, RevisionConflict, + RevisionCreationRequest, RevisionCreationResponse, RevisionDto, RevisionListItemDto, + RevisionListResponse, RevisionMetadataResponse, RevisionResponse, RevisionUpdateRequest, + ShareInvitationDto, ShareInvitationsResponse, ShareMembersResponse, ShareResponse, + ShareTargetType, ShareUrlDto, ShareUrlsResponse, SharedByMeResponse, SharedWithMeResponse, + SmallFileUploadMetadataRequest, SmallRevisionUploadMetadataRequest, SmallUploadResponse, + ThumbnailBlockListRequest, ThumbnailBlockListResponse, ThumbnailCreationRequest, ThumbnailDto, + TimelinePhotoListResponse, UpdatePermissionsRequest, VolumeCreationRequest, VolumeEventDto, + VolumeEventListResponse, VolumeTrashResponse, }; use crate::events::{DriveEvent, DriveEventScopeId}; use crate::node::{FileThumbnail, Node, NodeKind, RevisionState, Thumbnail, ThumbnailType}; @@ -78,6 +79,7 @@ use crate::sharing::{ Bookmark, ExternalInvitation, ExternalInvitationState, IncomingInvitation, MemberRole, PublicLink, ShareInvitation, ShareMember, }; +use crate::single_flight::SingleFlight; /// Content blocks are 4 MiB of plaintext each (C# `RevisionWriter.DefaultBlockSize`). const DEFAULT_BLOCK_SIZE: usize = 1 << 22; @@ -89,18 +91,50 @@ const SMALL_UPLOAD_SIZE_LIMIT: usize = 128 * 1024; /// `NodeOperations.MaximumBatchCount`). const MAX_BATCH_COUNT: usize = 150; -/// Content blocks a client keeps in memory at once, across every download it is -/// running. +/// Link-detail batches fetched at once, per volume, when enumerating nodes. +const MAX_CONCURRENT_DETAIL_FETCHES: usize = 4; + +/// Nodes decrypted at once within one batch of link details. /// -/// [`MAX_CONCURRENT_BLOCK_DOWNLOADS`] bounds one file; nothing bounded the host, -/// so N concurrent downloads cost N times that — a mount pinning a handful of -/// large files could hold hundreds of MiB of block buffers. This is the missing -/// global ceiling (TypeScript caps concurrent *files* instead, in -/// `internal/download/queue.ts`; capping blocks bounds the memory directly). +/// Each build runs its PGP work on the blocking pool (`crypto::decrypt_link_*`), +/// so this is what spreads a folder listing's per-node S2K derivations over +/// several cores; the author-key lookups it also overlaps are the smaller half. +const MAX_CONCURRENT_NODE_BUILDS: usize = 8; + +/// Content blocks a client keeps in memory at once, across every transfer it is +/// running — downloads and uploads alike. /// -/// Sized a little above one file's concurrency so a lone download still -/// saturates its pipeline and a second one is not starved. -const DEFAULT_MAX_INFLIGHT_BLOCKS: usize = 12; +/// [`MAX_CONCURRENT_BLOCK_DOWNLOADS`] and [`MAX_CONCURRENT_BLOCK_UPLOADS`] bound +/// one file; nothing bounded the host, so N concurrent transfers cost N times +/// that — a mount pinning a handful of large files could hold hundreds of MiB of +/// block buffers. This is the missing global ceiling (TypeScript caps concurrent +/// *files* instead, in `internal/download/queue.ts`; capping blocks bounds the +/// memory directly). +/// +/// Sized a little above one file's pipeline (an upload holds up to +/// `MAX_BUFFERED_UPLOAD_BLOCKS + MAX_CONCURRENT_BLOCK_UPLOADS + 1` permits) so a +/// lone transfer still saturates it and a second one is not starved. +const DEFAULT_MAX_INFLIGHT_BLOCKS: usize = 16; + +/// Content blocks uploaded concurrently within a single file. +/// +/// Matches the TypeScript SDK's `MAX_UPLOADING_BLOCKS` +/// (`internal/upload/streamUploader.ts`); C#'s shared transfer queue depth is 6 +/// (`ProtonDriveClient.DefaultDegreeOfBlockTransferParallelism`). +const MAX_CONCURRENT_BLOCK_UPLOADS: usize = 5; + +/// Encrypted blocks allowed to wait between the encryptor and the uploader. +/// +/// TypeScript buffers 15 (`MAX_BUFFERED_BLOCKS`); ours is much lower because a +/// buffered block also holds one of the client-wide in-flight block permits, +/// which are shared with downloads. +const MAX_BUFFERED_UPLOAD_BLOCKS: usize = 4; + +/// Attempts per block upload before the whole upload fails (TS +/// `MAX_BLOCK_UPLOAD_RETRIES`). Retries here are *above* the HTTP client's own +/// retry policy: they cover an expired upload token, which can only be fixed by +/// re-preparing the block. +const MAX_BLOCK_UPLOAD_ATTEMPTS: usize = 3; /// Trashed links requested per page when enumerating the trash (C# /// `VolumeOperations.TrashPageSize`). @@ -170,6 +204,15 @@ pub struct ProtonDriveClient { /// because it guards nothing else — `cache` stays free for every other /// caller — and the loser re-checks and finds the work already done. bootstrap: Arc, + /// In-flight [`get_node`](ProtonDriveClient::get_node) loads. Concurrent + /// lookups of the same node used to issue one link-details request and one + /// S2K each; now the first runs and the rest wait on it. + node_loads: Arc>>, + /// In-flight parent-key resolutions, keyed by the ancestor being resolved + /// *from* and the photos routing flag. Siblings enumerated concurrently all + /// want the same folder key, and resolving it walks (and decrypts) the whole + /// ancestor chain. + parent_key_loads: Arc>, } /// The one-time-per-account setup that [`ProtonDriveClient`] single-flights. @@ -275,6 +318,69 @@ struct BlockWriteResult { sha1_hex: String, } +/// One encrypted content block on its way to storage, as handed from the +/// encryptor to the uploader. +/// +/// It carries its own [`BlockCreationRequest`] because that is what re-minting +/// an expired upload token needs, and the client-wide in-flight block permit, +/// which is released only once the ciphertext has been stored. +struct PreparedBlock { + ciphertext: Bytes, + /// SHA-256 of the ciphertext — this block's manifest entry. + digest: [u8; 32], + request: BlockCreationRequest, + permit: OwnedSemaphorePermit, +} + +/// A thumbnail encrypted and ready to ride the first block-token request. +struct EncryptedThumbnail { + thumbnail_type: i32, + ciphertext: Bytes, + /// SHA-256 of the ciphertext — this thumbnail's manifest entry. + digest: [u8; 32], + request: ThumbnailCreationRequest, +} + +/// One blob to store: a content block or a thumbnail, with the target the +/// server issued for it. +struct UploadJob { + /// Block index, or `None` for a thumbnail (thumbnails contribute their + /// digest to the manifest up front, not through the upload result). + index: Option, + ciphertext: Bytes, + digest: [u8; 32], + /// The block's creation request, for re-minting an expired token. `None` for + /// thumbnails: the API only issues their tokens with the first block + /// request, so there is nothing to re-request them with (TS says the same). + request: Option, + target: BlockUploadTarget, + permit: Option, +} + +/// The ids a block upload needs to (re-)prepare a block. +/// +/// Owned rather than borrowed from the [`RevisionDraft`] so an upload future +/// never carries a lifetime — a borrowed one would make callers of +/// `upload_file_from` unspawnable (see `tests/spawnable.rs`). +#[derive(Clone)] +struct UploadContext { + address_id: AddressId, + volume_id: VolumeId, + link_id: LinkId, + revision_id: String, +} + +impl UploadContext { + fn from_draft(draft: &RevisionDraft) -> Self { + Self { + address_id: draft.address_id.clone(), + volume_id: draft.volume_id.clone(), + link_id: draft.link_id.clone(), + revision_id: draft.revision_id.clone(), + } + } +} + #[derive(Clone)] struct ShareKey { share_id: ShareId, @@ -312,6 +418,8 @@ impl ProtonDriveClient { block_slots: Arc::new(Semaphore::new(DEFAULT_MAX_INFLIGHT_BLOCKS)), small_file_upload: Arc::new(AtomicBool::new(false)), bootstrap: Arc::new(Bootstrap::default()), + node_loads: Arc::new(SingleFlight::default()), + parent_key_loads: Arc::new(SingleFlight::default()), } } @@ -338,6 +446,8 @@ impl ProtonDriveClient { block_slots: Arc::new(Semaphore::new(DEFAULT_MAX_INFLIGHT_BLOCKS)), small_file_upload: Arc::new(AtomicBool::new(false)), bootstrap: Arc::new(Bootstrap::default()), + node_loads: Arc::new(SingleFlight::default()), + parent_key_loads: Arc::new(SingleFlight::default()), } } @@ -378,13 +488,18 @@ impl ProtonDriveClient { } /// Cap the content blocks this client holds in memory at once, across every - /// download it is running. Defaults to [`DEFAULT_MAX_INFLIGHT_BLOCKS`]. + /// transfer it is running — downloads *and* uploads. Defaults to + /// [`DEFAULT_MAX_INFLIGHT_BLOCKS`]. /// /// A permit is held from the moment a block's fetch starts until its - /// plaintext is consumed, so `blocks * BLOCK_SIZE` is roughly the ceiling on - /// resident block memory — the knob to turn down on a memory-constrained - /// host, at the cost of download throughput. Must be non-zero; a zero cap - /// would deadlock every download, so it is clamped to 1. + /// plaintext is consumed (download), or from the moment its plaintext is + /// read until its ciphertext has been stored (upload), so + /// `blocks * BLOCK_SIZE` is roughly the ceiling on resident block memory — + /// the knob to turn down on a memory-constrained host, at the cost of + /// transfer throughput. Must be non-zero; a zero cap would deadlock every + /// transfer, so it is clamped to 1. Setting it to 1 serializes block + /// transfers entirely, which is what the throughput comparison in + /// `PERF_PLAN.md` uses to reproduce the pre-pipeline behavior. /// /// Applies to this client and every clone made from it *afterwards*; clones /// already taken keep the previous cap (they share the previous semaphore). @@ -403,8 +518,8 @@ impl ProtonDriveClient { self } - /// The global in-flight block permits, for [`RevisionReader`] and the - /// whole-file download path to acquire against. + /// The global in-flight block permits, for [`RevisionReader`], the + /// whole-file download path and the upload pipeline to acquire against. pub(crate) fn block_slots(&self) -> Arc { self.block_slots.clone() } @@ -443,12 +558,28 @@ impl ProtonDriveClient { timer.success(); return Ok(Some(info.node)); } + + // Concurrent callers asking for the same node share one load: the fetch + // *and* the node-key S2K behind it. + let client = self.clone(); + let target = uid.clone(); + let node = self + .node_loads + .run(uid.clone(), async move { client.load_node(&target).await }) + .await?; + + timer.success(); + Ok(node) + } + + /// The network half of [`get_node`](Self::get_node): link details, parent + /// key, decrypt. Runs behind [`ProtonDriveClient::node_loads`]. + async fn load_node(&self, uid: &NodeUid) -> Result> { let response = self .get_link_details(&uid.volume_id, std::slice::from_ref(&uid.link_id)) .await?; let Some(details) = response.links.into_iter().next() else { // Not found is a successful lookup, not a failure. - timer.success(); return Ok(None); }; @@ -458,7 +589,6 @@ impl ProtonDriveClient { let node = self .build_node(&uid.volume_id, &details, &parent_key, NodeDetail::Full) .await?; - timer.success(); Ok(Some(node)) } @@ -540,6 +670,20 @@ impl ProtonDriveClient { self.enumerate_nodes_detail(uids, NodeDetail::Light).await } + /// The shared body of [`enumerate_nodes`](Self::enumerate_nodes) and + /// [`enumerate_nodes_light`](Self::enumerate_nodes_light). + /// + /// Three things happen concurrently where they used to be strictly serial: + /// the per-volume batches are fetched [`MAX_CONCURRENT_DETAIL_FETCHES`] at a + /// time, and within a batch the nodes are decrypted + /// [`MAX_CONCURRENT_NODE_BUILDS`] at a time — which, now that the link crypto + /// runs on the blocking pool, spreads the per-node S2K over several cores. + /// Parent keys are still resolved *before* that fan-out, one per distinct + /// parent: siblings share a parent, and resolving concurrently would walk and + /// decrypt the same ancestor chain many times over. + /// + /// Order is preserved (`buffered`, not `buffer_unordered`), as is the + /// skip-and-warn behavior for a node whose parent key or own crypto fails. async fn enumerate_nodes_detail( &self, uids: &[NodeUid], @@ -548,29 +692,64 @@ impl ProtonDriveClient { let mut nodes = Vec::new(); for (volume_id, link_ids) in group_by_volume(uids) { - for chunk in link_ids.chunks(MAX_BATCH_COUNT) { - let details = self.get_link_details(&volume_id, chunk).await?; + let chunks: Vec> = link_ids + .chunks(MAX_BATCH_COUNT) + .map(<[LinkId]>::to_vec) + .collect(); + + let mut fetches = stream::iter(chunks.into_iter().map(|chunk| { + let client = self.clone(); + let volume_id = volume_id.clone(); + async move { client.get_link_details(&volume_id, &chunk).await } + })) + .buffered(MAX_CONCURRENT_DETAIL_FETCHES); + + while let Some(details) = fetches.try_next().await? { + let mut parent_keys: HashMap, PrivateKey> = HashMap::new(); for link_details in &details.links { - let parent_key = match self + let parent_id = link_details.link.parent_id.clone(); + if parent_keys.contains_key(&parent_id) { + continue; + } + match self .resolve_parent_key(&volume_id, &link_details.link) .await { - Ok(key) => key, + Ok(key) => { + parent_keys.insert(parent_id, key); + } Err(e) => { tracing::warn!(link_id = %link_details.link.id, error = %e, "skipping node: parent key unavailable"); - continue; - } - }; - match self - .build_node(&volume_id, link_details, &parent_key, detail) - .await - { - Ok(node) => nodes.push(node), - Err(e) => { - tracing::warn!(link_id = %link_details.link.id, error = %e, "skipping undecryptable node"); } } } + + let buildable = details.links.into_iter().filter_map(|link_details| { + let parent_key = parent_keys.get(&link_details.link.parent_id)?.clone(); + Some((link_details, parent_key)) + }); + let mut built = stream::iter(buildable.map(|(link_details, parent_key)| { + let client = self.clone(); + let volume_id = volume_id.clone(); + async move { + let link_id = link_details.link.id.clone(); + match client + .build_node(&volume_id, &link_details, &parent_key, detail) + .await + { + Ok(node) => Some(node), + Err(e) => { + tracing::warn!(link_id = %link_id, error = %e, "skipping undecryptable node"); + None + } + } + } + })) + .buffered(MAX_CONCURRENT_NODE_BUILDS); + + while let Some(node) = built.next().await { + nodes.extend(node); + } } } @@ -2390,76 +2569,191 @@ impl ProtonDriveClient { /// Move several nodes under a single destination parent in one batched /// request. Mirrors C# `ProtonDriveClient.MoveNodesAsync` / /// `NodeOperations.MoveMultipleAsync` (`PUT volumes/{vid}/links/move-multiple`, - /// note: no `v2/` prefix). Same-volume only — cross-volume is rejected, - /// matching the C# batch path, which also throws for differing volumes. Each - /// node's passphrase is rewrapped to the destination key and its name - /// re-encrypted + signed, exactly as the single [`move_node`]. Batched in - /// chunks of [`MAX_BATCH_COUNT`]; per-link failures surface via the aggregate - /// envelope. Live validation pending. - pub async fn move_nodes(&self, uids: &[NodeUid], new_parent: &NodeUid) -> Result<()> { + /// note: no `v2/` prefix — upstream still runs a per-node loop here and + /// FIXMEs the batch endpoint we already use). Same-volume only — cross-volume + /// is rejected, matching the C# batch path, which also throws for differing + /// volumes. Each node's passphrase is rewrapped to the destination key and its + /// name re-encrypted + signed, exactly as the single [`move_node`]. Batched in + /// chunks of [`MAX_BATCH_COUNT`]. Live validation pending. + /// + /// One outcome per input uid, in input order, like C#'s + /// `IReadOnlyDictionary>`: a node that fails — + /// cross-volume, unknown link, crypto, or a per-link error code in the batch + /// envelope — does not stop the others. The outer `Err` is reserved for + /// failures that make the whole call impossible (destination key, hash key, + /// signing address, link lookup). + pub async fn move_nodes( + &self, + uids: &[NodeUid], + new_parent: &NodeUid, + ) -> Result)>> { let mut timer = self.telemetry.start("move_nodes"); timer.attr("node_count", uids.len()); if uids.is_empty() { timer.success(); - return Ok(()); + return Ok(Vec::new()); } - for uid in uids { - if uid.volume_id != new_parent.volume_id { - return Err(ProtonError::invalid_operation( - "cross-volume move is not supported", - )); + + // Cross-volume nodes never reach the request — C# `MoveSingleAsync` throws + // for them, and here that throw is one node's outcome, not the call's. + let cross_volume: Vec = uids + .iter() + .map(|uid| uid.volume_id != new_parent.volume_id) + .collect(); + + // Each remaining link is moved once even if the caller listed it twice; + // every position sharing it reports the same outcome below. + let mut seen: HashSet = HashSet::with_capacity(uids.len()); + let mut targets: Vec = Vec::with_capacity(uids.len()); + for (uid, &skip) in uids.iter().zip(&cross_volume) { + if !skip && seen.insert(uid.link_id.clone()) { + targets.push(uid.link_id.clone()); } } - let dest_parent_key = self.folder_node_key(new_parent).await?; - let dest_hash_key = self.parent_hash_key(new_parent, &dest_parent_key).await?; - let (_address_id, email, signing_key) = self.membership_address().await?; + let mut link_outcomes: HashMap> = HashMap::with_capacity(targets.len()); + if !targets.is_empty() { + let dest_parent_key = self.folder_node_key(new_parent).await?; + let dest_hash_key = self.parent_hash_key(new_parent, &dest_parent_key).await?; + let (_address_id, email, signing_key) = self.membership_address().await?; - // Resolve every node's link details once (all share the destination - // volume), keyed by link id so each chunk can look its node up. - let link_ids: Vec = uids.iter().map(|u| u.link_id.clone()).collect(); - let mut links: std::collections::HashMap = - std::collections::HashMap::with_capacity(uids.len()); - for chunk in link_ids.chunks(MAX_BATCH_COUNT) { - let details = self.get_link_details(&new_parent.volume_id, chunk).await?; - for detail in details.links { - links.insert(detail.link.id.clone(), detail.link); + // Resolve every node's link details once (all share the destination + // volume), keyed by link id so each chunk can look its node up. + let mut links: HashMap = HashMap::with_capacity(targets.len()); + for chunk in targets.chunks(MAX_BATCH_COUNT) { + let details = self.get_link_details(&new_parent.volume_id, chunk).await?; + for detail in details.links { + links.insert(detail.link.id.clone(), detail.link); + } + } + + for chunk in targets.chunks(MAX_BATCH_COUNT) { + let mut items = Vec::with_capacity(chunk.len()); + // The links this request actually carries, so the per-link + // responses can be routed back. + let mut sent: Vec = Vec::with_capacity(chunk.len()); + for link_id in chunk { + let uid = NodeUid::new(new_parent.volume_id.clone(), link_id.clone()); + let Some(link) = links.get(link_id) else { + link_outcomes.insert( + link_id.clone(), + Err(ProtonError::invalid_operation(format!( + "node {uid} not found" + ))), + ); + continue; + }; + let parts = match self + .build_move_parts( + &uid, + link, + &dest_parent_key, + &dest_hash_key, + &signing_key, + ) + .await + { + Ok(parts) => parts, + Err(e) => { + link_outcomes.insert(link_id.clone(), Err(e)); + continue; + } + }; + items.push(MoveMultipleLinksItem { + link_id: link_id.clone(), + name: parts.encrypted_name, + passphrase: parts.passphrase, + name_hash: parts.name_hash, + original_hash: parts.original_hash, + // The rewrap preserves the plaintext; the existing detached + // passphrase signature stays valid, so none is re-sent (C# + // omits it for non-anonymous nodes). + passphrase_signature: None, + }); + sent.push(link_id.clone()); + } + if items.is_empty() { + continue; + } + + let request = MoveMultipleLinksRequest { + parent_link_id: new_parent.link_id.clone(), + links: items, + name_signature_email: email.clone(), + signature_email: None, + }; + let path = format!("volumes/{}/links/move-multiple", new_parent.volume_id); + let response: AggregateLinksResponse = match self.http.put(&path, &request).await { + Ok(response) => response, + Err(e) => { + // The chunk failed as a whole; charge it to every node it + // carried and keep going with the remaining chunks, so one + // bad chunk cannot mask the rest (C#'s per-node loop never + // stops early either). + let message = e.to_string(); + for link_id in sent { + link_outcomes.insert( + link_id, + Err(ProtonError::invalid_operation(message.clone())), + ); + } + continue; + } + }; + + let mut per_link: HashMap> = + aggregate_outcomes(response).into_iter().collect(); + for link_id in sent { + let outcome = per_link.remove(&link_id).unwrap_or_else(|| { + Err(ProtonError::invalid_operation(format!( + "move returned no response for link {link_id}" + ))) + }); + link_outcomes.insert(link_id, outcome); + } } } - for chunk in uids.chunks(MAX_BATCH_COUNT) { - let mut items = Vec::with_capacity(chunk.len()); - for uid in chunk { - let link = links.get(&uid.link_id).ok_or_else(|| { - ProtonError::invalid_operation(format!("node {uid} not found")) - })?; - let parts = self - .build_move_parts(uid, link, &dest_parent_key, &dest_hash_key, &signing_key) - .await?; - items.push(MoveMultipleLinksItem { - link_id: uid.link_id.clone(), - name: parts.encrypted_name, - passphrase: parts.passphrase, - name_hash: parts.name_hash, - original_hash: parts.original_hash, - // The rewrap preserves the plaintext; the existing detached - // passphrase signature stays valid, so none is re-sent (C# - // omits it for non-anonymous nodes). - passphrase_signature: None, - }); - } - let request = MoveMultipleLinksRequest { - parent_link_id: new_parent.link_id.clone(), - links: items, - name_signature_email: email.clone(), - signature_email: None, - }; - let path = format!("volumes/{}/links/move-multiple", new_parent.volume_id); - let response: AggregateLinksResponse = self.http.put(&path, &request).await?; - check_aggregate("move", response)?; - } + // A repeated uid cannot take the owned `Result` twice, so keep each + // link's error text to rebuild the outcome for its later positions. + let repeats: HashMap> = link_outcomes + .iter() + .map(|(link_id, outcome)| { + ( + link_id.clone(), + outcome.as_ref().err().map(|e| e.to_string()), + ) + }) + .collect(); + + let results: Vec<(NodeUid, Result<()>)> = uids + .iter() + .zip(&cross_volume) + .map(|(uid, &skip)| { + let outcome = if skip { + Err(ProtonError::invalid_operation( + "cross-volume move is not supported", + )) + } else if let Some(outcome) = link_outcomes.remove(&uid.link_id) { + outcome + } else { + match repeats.get(&uid.link_id) { + Some(None) => Ok(()), + Some(Some(message)) => Err(ProtonError::invalid_operation(message.clone())), + None => Err(ProtonError::invalid_operation(format!( + "move produced no outcome for node {uid}" + ))), + } + }; + (uid.clone(), outcome) + }) + .collect(); + timer.attr( + "failed_count", + results.iter().filter(|(_, r)| r.is_err()).count(), + ); timer.success(); - Ok(()) + Ok(results) } /// Build the per-node move crypto shared by [`move_node`] and [`move_nodes`]: @@ -2651,6 +2945,45 @@ impl ProtonDriveClient { pub async fn enumerate_shared_by_me_node_uids(&self) -> Result> { let mut timer = self.telemetry.start("enumerate_shared_by_me_node_uids"); let volume_id = self.main_volume_id().await?; + let uids = self.page_shared_by_me(&volume_id).await?; + timer.success(); + Ok(uids) + } + + /// The photos I have shared with others, as [`NodeUid`]s — the same listing + /// as [`enumerate_shared_by_me_node_uids`](Self::enumerate_shared_by_me_node_uids) + /// but resolved against the photos volume, mirroring C# + /// `ProtonPhotosClient.EnumerateSharedNodeUidsAsync` (which passes + /// `VolumeOperations.TryGetPhotosVolumeIdAsync` as the volume resolver). + /// Empty when the account has no photos volume — C# yields nothing there + /// rather than erroring. Exposed through + /// [`ProtonPhotosClient`](crate::ProtonPhotosClient). + pub(crate) async fn enumerate_photos_shared_by_me_node_uids(&self) -> Result> { + let mut timer = self + .telemetry + .start("enumerate_photos_shared_by_me_node_uids"); + if !self.ensure_photos().await? { + timer.success(); + return Ok(Vec::new()); + } + let volume_id = self + .cache + .lock() + .await + .photos_root + .clone() + .expect("ensure_photos populated the photos root") + .volume_id; + + let uids = self.page_shared_by_me(&volume_id).await?; + timer.success(); + Ok(uids) + } + + /// Page `GET drive/v2/volumes/{vid}/shares` on its `AnchorID` cursor for one + /// volume. Mirrors the loop in C# `SharingOperations.EnumerateSharedNodeUidsAsync` + /// once the volume resolver has run. + async fn page_shared_by_me(&self, volume_id: &VolumeId) -> Result> { let mut uids = Vec::new(); let mut anchor: Option = None; @@ -2673,7 +3006,6 @@ impl ProtonDriveClient { } } - timer.success(); Ok(uids) } @@ -4040,14 +4372,28 @@ impl ProtonDriveClient { /// Encrypt, sign, verify and upload every content block of a draft revision, /// accumulating the content manifest and extended-attribute metadata. /// - /// Thumbnails (if any) are uploaded first, in `ThumbnailType` order, and - /// their ciphertext digests lead the manifest — mirroring C# - /// `RevisionWriter.UploadBlocksAsync` (thumbnails then content blocks) and - /// the download path's manifest ordering. Then `reader` is streamed one - /// [`DEFAULT_BLOCK_SIZE`] block at a time: only a single plaintext block - /// (plus its ciphertext) is held in memory, and the SHA-1 digest and - /// total-size counter are folded incrementally. An empty reader yields zero - /// content blocks (an empty file). + /// Two halves run concurrently, mirroring the TypeScript SDK's + /// `internal/upload/streamUploader.ts`: an **encryptor** reads the plaintext + /// block by block and hands encrypted blocks to an **uploader** over a + /// bounded channel; the uploader asks for upload tokens for every block that + /// is ready (one `POST blocks` per batch, thumbnails riding the first one) + /// and pushes up to [`MAX_CONCURRENT_BLOCK_UPLOADS`] of them at a time. + /// + /// Both halves stay on *this* task: `reader` is only `Read + Send`, never + /// `'static` (`upload_file` hands in a `&[u8]`), so it cannot move into a + /// spawned task. Only the CPU-bound per-block crypto is offloaded, via + /// `spawn_blocking`. + /// + /// Unlike TS, which keeps a sliding window, a batch's tokens are requested + /// once the previous batch has finished storing. That costs one round-trip + /// per batch of overlap and keeps the ordering bookkeeping trivial. + /// + /// Thumbnail digests lead the manifest in `ThumbnailType` order and content + /// digests follow in block-index order — the layout the download path + /// verifies, unchanged even though blocks now reach storage out of order + /// (each block carries its index on the wire). SHA-1, block sizes and the + /// total-size counter are folded by the encryptor, in order. An empty reader + /// yields zero content blocks (an empty file). async fn write_blocks( &self, draft: &RevisionDraft, @@ -4064,123 +4410,244 @@ impl ProtonDriveClient { ) .await?; - let mut manifest = Vec::new(); - - // Thumbnail digests lead the manifest, ordered by thumbnail type to - // match the download path's `sort_by_key(thumbnail_type)`. + // Thumbnails are encrypted up front, in type order: the API only issues + // thumbnail upload tokens with the *first* block-token request (TS + // `requestAndInitiateUpload`), and they are tiny enough that encrypting + // them inline costs nothing. thumbnails.sort_by_key(|t| t.thumbnail_type); - for thumbnail in &thumbnails { - let digest = self.upload_thumbnail(draft, thumbnail).await?; - manifest.extend_from_slice(&digest); + let encrypted_thumbnails = thumbnails + .iter() + .map(|thumbnail| encrypt_thumbnail_block(draft, thumbnail)) + .collect::>>()?; + let thumbnail_digests: Vec<[u8; 32]> = + encrypted_thumbnails.iter().map(|t| t.digest).collect(); + + let (tx, rx) = mpsc::channel(MAX_BUFFERED_UPLOAD_BLOCKS); + let (written, content_digests) = tokio::try_join!( + self.encrypt_content_blocks(draft, &mut reader, &verification_code, tx), + self.upload_prepared_blocks(draft, rx, encrypted_thumbnails), + )?; + let (block_sizes, total_size, sha1_hex) = written; + + if content_digests.len() != block_sizes.len() { + return Err(ProtonError::invalid_operation(format!( + "upload stored {} of {} content blocks", + content_digests.len(), + block_sizes.len() + ))); } + Ok(BlockWriteResult { + manifest: assemble_manifest(&thumbnail_digests, content_digests), + block_sizes, + total_size, + sha1_hex, + }) + } + + /// Read `reader` one [`DEFAULT_BLOCK_SIZE`] block at a time, encrypt and sign + /// each block off the runtime, and hand it to the uploader over `tx`. + /// + /// Returns the per-block plaintext sizes, the total plaintext size and the + /// hex SHA-1 of the whole plaintext — all folded in block order, which is why + /// this half stays sequential. TS `encryptBlocks` is sequential for the same + /// reason and relies on buffering, not parallel encryption, to hide the cost. + async fn encrypt_content_blocks( + &self, + draft: &RevisionDraft, + reader: &mut R, + verification_code: &[u8], + tx: mpsc::Sender, + ) -> Result<(Vec, i64, String)> { let mut block_sizes = Vec::new(); let mut sha1 = Sha1::new(); let mut total_size: i64 = 0; - let mut buf = vec![0u8; DEFAULT_BLOCK_SIZE]; // Block indices are 1-based (C# `blockNumber = i + 1`). let mut index = 1_i32; loop { - let n = read_full_block(&mut reader, &mut buf)?; + // The client-wide in-flight block permit, held until this block's + // ciphertext has been stored. + let permit = + self.block_slots().acquire_owned().await.map_err(|e| { + ProtonError::invalid_operation(format!("block slots closed: {e}")) + })?; + + // A fresh buffer per block rather than one reused across the loop: + // the plaintext is moved into the blocking encrypt task, and copying + // 4 MiB out of a shared buffer would cost as much as allocating. + let mut buf = vec![0u8; DEFAULT_BLOCK_SIZE]; + let n = read_full_block(reader, &mut buf)?; if n == 0 { break; } - let chunk = &buf[..n]; + buf.truncate(n); - sha1.update(chunk); + sha1.update(&buf); total_size += n as i64; - - let ciphertext = draft.content_key.encrypt_block(chunk)?; - let digest = Sha256::digest(&ciphertext); - let token = verification_token(&verification_code, &ciphertext); - - // Detached signature over the plaintext, then encrypted to the node key. - let plaintext_signature = draft.signing_key.sign_detached(chunk)?; - let encrypted_signature = draft.node_key.encrypt(plaintext_signature.as_bytes())?; - - let prepare = BlockUploadPreparationRequest { - address_id: draft.address_id.clone(), - volume_id: draft.volume_id.clone(), - link_id: draft.link_id.clone(), - revision_id: draft.revision_id.clone(), - blocks: vec![BlockCreationRequest { - index, - size: ciphertext.len() as i32, - encrypted_signature, - hash: BASE64.encode(digest), - verifier: BlockVerifier { - token: BASE64.encode(&token), - }, - }], - thumbnails: Vec::new(), - }; - let prepared: BlockUploadPreparationResponse = - self.http.post("blocks", &prepare).await?; - let target = prepared.upload_targets.into_iter().next().ok_or_else(|| { - ProtonError::invalid_operation("block upload preparation returned no target") - })?; - - self.http - .post_storage_blob(&target.bare_url, &target.token, ciphertext) - .await?; - - manifest.extend_from_slice(&digest); block_sizes.push(n as i32); + + let content_key = draft.content_key.clone(); + let node_key = draft.node_key.clone(); + let signing_key = draft.signing_key.clone(); + let code = verification_code.to_vec(); + let (ciphertext, digest, request) = tokio::task::spawn_blocking(move || { + encrypt_content_block(&content_key, &node_key, &signing_key, &code, index, &buf) + }) + .await + .map_err(|e| { + ProtonError::invalid_operation(format!("block encrypt task failed: {e}")) + })??; + + let prepared = PreparedBlock { + ciphertext, + digest, + request, + permit, + }; + if tx.send(prepared).await.is_err() { + // The uploader is gone, which means it failed: `try_join!` will + // surface its error rather than anything we could say here. + break; + } index += 1; } - Ok(BlockWriteResult { - manifest, - block_sizes, - total_size, - sha1_hex: hex::encode(sha1.finalize()), - }) + Ok((block_sizes, total_size, hex::encode(sha1.finalize()))) } - /// Encrypt, hash and upload a single thumbnail block, returning its - /// ciphertext SHA-256 digest (a manifest entry). + /// Take encrypted blocks off `rx`, request upload targets for each batch and + /// store the ciphertext, [`MAX_CONCURRENT_BLOCK_UPLOADS`] blocks at a time. /// - /// Mirrors C# `BlockUploader.UploadThumbnailAsync`: the thumbnail is - /// encrypt-and-inline-signed under the content key, prepared as its own - /// upload request (no content blocks), and uploaded to the returned - /// thumbnail target. - async fn upload_thumbnail( + /// Returns `(block index, ciphertext digest)` per stored content block, in + /// completion order — the caller sorts. Thumbnails ride the first batch and + /// contribute no entries; their digests are already known to the caller. + async fn upload_prepared_blocks( &self, draft: &RevisionDraft, - thumbnail: &Thumbnail, - ) -> Result<[u8; 32]> { - let ciphertext = draft - .content_key - .encrypt_thumbnail(&draft.signing_key, &thumbnail.content)?; - let digest = Sha256::digest(&ciphertext); + mut rx: mpsc::Receiver, + thumbnails: Vec, + ) -> Result> { + let context = UploadContext::from_draft(draft); + // Both are local to this one file's upload: a timeout downshift must not + // leak into other transfers (TS `limitUploadCapacity`). + let limiter = Arc::new(Semaphore::new(MAX_CONCURRENT_BLOCK_UPLOADS)); + let downshifted = Arc::new(AtomicBool::new(false)); - let prepare = BlockUploadPreparationRequest { - address_id: draft.address_id.clone(), - volume_id: draft.volume_id.clone(), - link_id: draft.link_id.clone(), - revision_id: draft.revision_id.clone(), - blocks: Vec::new(), - thumbnails: vec![ThumbnailCreationRequest { - size: ciphertext.len() as i32, - thumbnail_type: thumbnail.thumbnail_type.as_i32(), - hash: BASE64.encode(digest), - }], - }; - let prepared: BlockUploadPreparationResponse = self.http.post("blocks", &prepare).await?; - let target = prepared - .thumbnail_upload_targets - .into_iter() - .next() - .ok_or_else(|| { - ProtonError::invalid_operation("thumbnail upload preparation returned no target") - })?; + let mut pending_thumbnails = thumbnails; + let mut digests = Vec::new(); + let mut batch: Vec = Vec::with_capacity(MAX_BUFFERED_UPLOAD_BLOCKS + 1); - self.http - .post_storage_blob(&target.bare_url, &target.token, ciphertext) + loop { + // Blocks the encryptor has finished so far, however many that is: + // one token request per batch instead of one per block. + if rx + .recv_many(&mut batch, MAX_BUFFERED_UPLOAD_BLOCKS + 1) + .await + == 0 + { + break; + } + + let stored = self + .upload_batch( + &context, + &limiter, + &downshifted, + std::mem::take(&mut batch), + std::mem::take(&mut pending_thumbnails), + ) + .await?; + digests.extend(stored); + } + + // A file with thumbnails but no content at all still has to ship them, + // and they never got a batch to ride along with. + if !pending_thumbnails.is_empty() { + self.upload_batch( + &context, + &limiter, + &downshifted, + Vec::new(), + pending_thumbnails, + ) .await?; + } - Ok(digest.into()) + Ok(digests) + } + + /// Request upload targets for one batch of encrypted blocks (plus any + /// thumbnails still waiting for their first request) and store them, + /// [`MAX_CONCURRENT_BLOCK_UPLOADS`] at a time. + async fn upload_batch( + &self, + context: &UploadContext, + limiter: &Arc, + downshifted: &Arc, + batch: Vec, + thumbnails: Vec, + ) -> Result> { + let requests: Vec = + batch.iter().map(|block| block.request.clone()).collect(); + let thumbnail_requests: Vec = thumbnails + .iter() + .map(|thumbnail| thumbnail.request.clone()) + .collect(); + let prepared = + request_upload_targets(&self.http, context, requests, thumbnail_requests).await?; + let mut block_targets = prepared.upload_targets; + let mut thumbnail_targets = prepared.thumbnail_upload_targets; + + let mut jobs = Vec::with_capacity(batch.len() + thumbnails.len()); + for thumbnail in thumbnails { + let target = take_thumbnail_target(&mut thumbnail_targets, thumbnail.thumbnail_type) + .ok_or_else(|| { + ProtonError::invalid_operation(format!( + "thumbnail upload preparation returned no target for type {}", + thumbnail.thumbnail_type + )) + })?; + jobs.push(UploadJob { + index: None, + ciphertext: thumbnail.ciphertext, + digest: thumbnail.digest, + request: None, + target, + permit: None, + }); + } + for block in batch { + let index = block.request.index; + let target = take_block_target(&mut block_targets, index).ok_or_else(|| { + ProtonError::invalid_operation(format!( + "block upload preparation returned no target for block {index}" + )) + })?; + jobs.push(UploadJob { + index: Some(index), + ciphertext: block.ciphertext, + digest: block.digest, + request: Some(block.request), + target, + permit: Some(block.permit), + }); + } + + // Each job owns everything it needs; nothing borrows the draft, so this + // future stays `tokio::spawn`-able for callers (see `tests/spawnable.rs`). + let stored: Vec> = stream::iter(jobs.into_iter().map(|job| { + let http = self.http.clone(); + let context = context.clone(); + let limiter = limiter.clone(); + let downshifted = downshifted.clone(); + async move { upload_one_block(&http, &context, job, &limiter, &downshifted).await } + })) + .buffer_unordered(MAX_CONCURRENT_BLOCK_UPLOADS) + .try_collect() + .await?; + + Ok(stored.into_iter().flatten().collect()) } /// Seal a draft revision: PUT a signed content manifest plus encrypted + @@ -4986,12 +5453,49 @@ impl ProtonDriveClient { volume_id: &VolumeId, link: &LinkDto, for_photos: bool, + ) -> Result { + // An already-decrypted parent key needs neither a walk nor a queue. + if let Some(parent_id) = &link.parent_id { + let uid = NodeUid::new(volume_id.clone(), parent_id.clone()); + if let Some(key) = self.cache.lock().await.folder_keys.get(&uid) { + return Ok(key.clone()); + } + } + + // Siblings resolved concurrently all want this same key, and getting it + // means walking and decrypting the whole ancestor chain — so one of them + // does it and the rest wait. Parentless links key on their own id: what + // they resolve to is that root's share key. + let target = NodeUid::new( + volume_id.clone(), + link.parent_id.clone().unwrap_or_else(|| link.id.clone()), + ); + let client = self.clone(); + let volume_id = volume_id.clone(); + let link = link.clone(); + self.parent_key_loads + .run((target, for_photos), async move { + client + .resolve_parent_key_walk(&volume_id, &link, for_photos) + .await + }) + .await + } + + /// The ancestor walk behind [`resolve_parent_key_ctx`]: climb to the nearest + /// decrypted key (or the root share key) and decrypt downward from there, + /// caching every folder key on the way. + async fn resolve_parent_key_walk( + &self, + volume_id: &VolumeId, + link: &LinkDto, + for_photos: bool, ) -> Result { let mut ancestry: Vec = Vec::new(); let mut current = link.parent_id.clone(); let mut base_key: Option = None; - while let Some(parent_id) = current { + while let Some(parent_id) = current.take() { let uid = NodeUid::new(volume_id.clone(), parent_id.clone()); if let Some(key) = self.cache.lock().await.folder_keys.get(&uid) { @@ -4999,18 +5503,29 @@ impl ProtonDriveClient { break; } + // Walk as far up as the entity cache can take us for free, then + // fetch that whole run of ancestors in one request instead of one + // request per level. + let run = self.cached_ancestor_run(volume_id, &parent_id).await; let details = self - .get_link_details_ctx(volume_id, std::slice::from_ref(&parent_id), for_photos) + .get_link_details_ctx(volume_id, &run, for_photos) .await?; - let ancestor = details + let mut by_id: HashMap = details .links .into_iter() - .next() - .ok_or_else(|| ProtonError::invalid_operation(format!("ancestor {uid} not found")))? - .link; + .map(|details| (details.link.id.clone(), details.link)) + .collect(); - current = ancestor.parent_id.clone(); - ancestry.push(ancestor); + for ancestor_id in &run { + let ancestor = by_id.remove(ancestor_id).ok_or_else(|| { + ProtonError::invalid_operation(format!( + "ancestor {} not found", + NodeUid::new(volume_id.clone(), ancestor_id.clone()) + )) + })?; + current = ancestor.parent_id.clone(); + ancestry.push(ancestor); + } } // Start from the resolved base (cached ancestor key, or the share key @@ -5043,6 +5558,45 @@ impl ProtonDriveClient { Ok(key) } + /// The run of ancestors starting at `first` that the entity cache can name + /// without touching the network, so they can be fetched in one batch. + /// + /// The chain is only walked for *ids* — the cached node carries its + /// `parent_uid`, never the encrypted material the key walk needs. It stops + /// at the first ancestor the cache does not know, at a root, or at one whose + /// folder key is already decrypted (the caller stops there anyway), so the + /// returned run never contains an ancestor that would not have been fetched. + /// With a cold entity cache it degenerates to `[first]` — exactly the + /// one-level-at-a-time walk this replaced. + async fn cached_ancestor_run(&self, volume_id: &VolumeId, first: &LinkId) -> Vec { + let mut run = vec![first.clone()]; + + let mut current = first.clone(); + while run.len() < MAX_BATCH_COUNT { + let uid = NodeUid::new(volume_id.clone(), current.clone()); + let Ok(Some(info)) = self.entities.try_get_node(&uid).await else { + break; + }; + let Some(parent_uid) = info.node.parent_uid else { + break; + }; + if self + .cache + .lock() + .await + .folder_keys + .get(&parent_uid) + .is_some() + { + break; + } + current = parent_uid.link_id; + run.push(current.clone()); + } + + run + } + async fn get_link_details( &self, volume_id: &VolumeId, @@ -5583,12 +6137,40 @@ struct MoveParts { original_hash: String, } -fn check_aggregate(op: &str, response: AggregateLinksResponse) -> Result<()> { - let failures: Vec = response +/// Split a `1001 MultipleResponses` batch envelope into one outcome per link, +/// in the order the API listed them. A non-success per-link code becomes a +/// [`ProtonApiError`] carrying that code — callers that need to branch on +/// *why* a link failed (C# `MoveNodesAsync`'s per-node result) get the machine +/// -readable code, not just a message. `http_status` is the batch request's own +/// 200: the transport succeeded, the individual link did not. +fn aggregate_outcomes(response: AggregateLinksResponse) -> Vec<(LinkId, Result<()>)> { + response .responses - .iter() - .filter(|pair| !pair.response.is_success()) - .map(|pair| format!("{} ({:?})", pair.link_id, pair.response.code)) + .into_iter() + .map(|pair| { + let outcome = if pair.response.is_success() { + Ok(()) + } else { + Err(ProtonError::Api(ProtonApiError { + code: pair.response.code, + http_status: 200, + message: pair.response.error_message.unwrap_or_default(), + details: pair.response.details, + })) + }; + (pair.link_id, outcome) + }) + .collect() +} + +fn check_aggregate(op: &str, response: AggregateLinksResponse) -> Result<()> { + let failures: Vec = aggregate_outcomes(response) + .into_iter() + .filter_map(|(link_id, outcome)| match outcome { + Ok(()) => None, + Err(ProtonError::Api(e)) => Some(format!("{link_id} ({:?})", e.code)), + Err(e) => Some(format!("{link_id} ({e})")), + }) .collect(); if failures.is_empty() { Ok(()) @@ -5642,17 +6224,299 @@ fn verification_token(code: &[u8], ciphertext: &[u8]) -> Vec { .collect() } +/// Assemble a revision's content manifest: thumbnail digests in `ThumbnailType` +/// order, then content-block digests in block-index order. +/// +/// That layout is what the manifest signature covers and what the download path +/// re-derives when it verifies a revision, so it has to hold regardless of the +/// order blocks actually reached storage in — which, with the upload pipelined, +/// is no longer the order they were read in. +fn assemble_manifest(thumbnails: &[[u8; 32]], mut blocks: Vec<(i32, [u8; 32])>) -> Vec { + blocks.sort_by_key(|(index, _)| *index); + + let mut manifest = Vec::with_capacity((thumbnails.len() + blocks.len()) * 32); + for digest in thumbnails { + manifest.extend_from_slice(digest); + } + for (_, digest) in &blocks { + manifest.extend_from_slice(digest); + } + manifest +} + +/// Encrypt, digest and sign one content block, producing everything the upload +/// and the manifest need. +/// +/// CPU-bound (PGP over 4 MiB plus a signature), so callers run it on the +/// blocking pool — the download path offloads its counterpart the same way +/// (`revision::decrypt_block_blocking`). +fn encrypt_content_block( + content_key: &ContentKey, + node_key: &PrivateKey, + signing_key: &PrivateKey, + verification_code: &[u8], + index: i32, + plaintext: &[u8], +) -> Result<(Bytes, [u8; 32], BlockCreationRequest)> { + let ciphertext = content_key.encrypt_block(plaintext)?; + let digest: [u8; 32] = Sha256::digest(&ciphertext).into(); + let token = verification_token(verification_code, &ciphertext); + + // Detached signature over the plaintext, then encrypted to the node key. + let plaintext_signature = signing_key.sign_detached(plaintext)?; + let encrypted_signature = node_key.encrypt(plaintext_signature.as_bytes())?; + + let request = BlockCreationRequest { + index, + size: ciphertext.len() as i32, + encrypted_signature, + hash: BASE64.encode(digest), + verifier: BlockVerifier { + token: BASE64.encode(&token), + }, + }; + + Ok((Bytes::from(ciphertext), digest, request)) +} + +/// Encrypt and inline-sign a thumbnail under the content key, ready to ride the +/// first block-token request. Mirrors C# `BlockUploader.UploadThumbnailAsync`'s +/// crypto half. +fn encrypt_thumbnail_block( + draft: &RevisionDraft, + thumbnail: &Thumbnail, +) -> Result { + let ciphertext = draft + .content_key + .encrypt_thumbnail(&draft.signing_key, &thumbnail.content)?; + let digest: [u8; 32] = Sha256::digest(&ciphertext).into(); + let thumbnail_type = thumbnail.thumbnail_type.as_i32(); + + Ok(EncryptedThumbnail { + thumbnail_type, + request: ThumbnailCreationRequest { + size: ciphertext.len() as i32, + thumbnail_type, + hash: BASE64.encode(digest), + }, + ciphertext: Bytes::from(ciphertext), + digest, + }) +} + +/// `POST blocks`: ask for upload targets for a batch of encrypted blocks, plus +/// this upload's thumbnails (only ever on its first request). +async fn request_upload_targets( + http: &ApiHttpClient, + context: &UploadContext, + blocks: Vec, + thumbnails: Vec, +) -> Result { + let request = BlockUploadPreparationRequest { + address_id: context.address_id.clone(), + volume_id: context.volume_id.clone(), + link_id: context.link_id.clone(), + revision_id: context.revision_id.clone(), + blocks, + thumbnails, + }; + http.post("blocks", &request).await +} + +/// Take the target the server issued for content block `index`, falling back to +/// request order when the response does not echo indices. +fn take_block_target( + targets: &mut Vec, + index: i32, +) -> Option { + let position = targets + .iter() + .position(|target| target.index == Some(index)) + .or_else(|| targets.iter().position(|target| target.index.is_none()))?; + Some(targets.remove(position)) +} + +/// As [`take_block_target`], for the thumbnail of a given type. +fn take_thumbnail_target( + targets: &mut Vec, + thumbnail_type: i32, +) -> Option { + let position = targets + .iter() + .position(|target| target.thumbnail_type == Some(thumbnail_type)) + .or_else(|| { + targets + .iter() + .position(|target| target.thumbnail_type.is_none()) + })?; + Some(targets.remove(position)) +} + +/// Store one encrypted blob, returning `(index, digest)` for a content block and +/// `None` for a thumbnail. +/// +/// Mirrors TS `streamUploader.uploadBlock`: retry up to +/// [`MAX_BLOCK_UPLOAD_ATTEMPTS`] times, re-preparing the block when its upload +/// token has expired and serializing the whole file's uploads after a timeout. +/// This sits *above* the HTTP client's own retry policy, which already covers +/// transport failures, 5xx and 429 — what it cannot do is mint a new token, +/// which needs the block's signature and verifier. +async fn upload_one_block( + http: &ApiHttpClient, + context: &UploadContext, + job: UploadJob, + limiter: &Semaphore, + downshifted: &AtomicBool, +) -> Result> { + let UploadJob { + index, + ciphertext, + digest, + request, + mut target, + permit: _permit, + } = job; + + let mut attempt = 0_usize; + loop { + attempt += 1; + + let outcome = { + // Once downshifted, a block takes the entire allowance, which is how + // the upload serializes itself for the rest of the file (TS + // `limitUploadCapacity` waits for every other block instead). + let wanted = if downshifted.load(Ordering::Relaxed) { + MAX_CONCURRENT_BLOCK_UPLOADS as u32 + } else { + 1 + }; + let _slot = limiter.acquire_many(wanted).await.map_err(|e| { + ProtonError::invalid_operation(format!("upload limiter closed: {e}")) + })?; + http.post_storage_blob(&target.bare_url, &target.token, ciphertext.clone()) + .await + }; + + let error = match outcome { + Ok(()) => return Ok(index.map(|index| (index, digest))), + Err(error) => error, + }; + if attempt >= MAX_BLOCK_UPLOAD_ATTEMPTS { + return Err(error); + } + + if is_upload_timeout(&error) { + downshifted.store(true, Ordering::Relaxed); + } else if is_expired_upload_token(&error) + && let Some(request) = request.as_ref() + { + // Retrying the same URL can only fail the same way, so re-prepare + // the block and aim at the fresh target. + let prepared = + request_upload_targets(http, context, vec![request.clone()], Vec::new()).await?; + target = prepared.upload_targets.into_iter().next().ok_or_else(|| { + ProtonError::invalid_operation("block upload preparation returned no target") + })?; + } + + tracing::warn!( + block = ?index, + attempt, + error = %error, + "block upload failed, retrying" + ); + } +} + +/// The storage host no longer knows this upload token — the block has to be +/// re-prepared before it can be retried (TS `uploadBlock`'s `NOT_FOUND` branch). +fn is_expired_upload_token(error: &ProtonError) -> bool { + matches!( + error, + ProtonError::Api(e) if e.http_status == 404 || matches!(e.code, ResponseCode::DoesNotExist) + ) +} + +/// The upload ran out of time rather than failing outright, which TS treats as a +/// signal that the link cannot carry the current concurrency. +fn is_upload_timeout(error: &ProtonError) -> bool { + matches!(error, ProtonError::Transport(e) if e.is_timeout()) +} + #[cfg(test)] mod tests { use super::{ - DriveCache, DriveEvent, FOLDER_KEY_CACHE_CAP, alternate_names, epoch_to_iso8601, - is_listable_revision_state, join_node_path, path_segments, small_upload_applicable, - to_drive_event, + DriveCache, DriveEvent, FOLDER_KEY_CACHE_CAP, aggregate_outcomes, alternate_names, + assemble_manifest, check_aggregate, epoch_to_iso8601, is_expired_upload_token, + is_listable_revision_state, is_upload_timeout, join_node_path, path_segments, + small_upload_applicable, take_block_target, take_thumbnail_target, to_drive_event, + }; + use crate::dtos::{ + AggregateLinksResponse, BlockUploadTarget, LinkIdResponsePair, VolumeEventDto, + VolumeEventLinkDto, }; - use crate::dtos::{VolumeEventDto, VolumeEventLinkDto}; use crate::node::RevisionState; + use proton_sdk::api::{ApiResponse, ResponseCode}; + use proton_sdk::error::{ProtonApiError, ProtonError}; use proton_sdk::ids::{DriveEventId, LinkId, VolumeId}; + /// A batch envelope with one successful link and one that failed with `code`. + fn mixed_aggregate(code: ResponseCode) -> AggregateLinksResponse { + AggregateLinksResponse { + responses: vec![ + LinkIdResponsePair { + link_id: LinkId::new("ok-link"), + response: ApiResponse { + code: ResponseCode::Success, + error_message: None, + details: None, + }, + }, + LinkIdResponsePair { + link_id: LinkId::new("bad-link"), + response: ApiResponse { + code, + error_message: Some("nope".to_string()), + details: None, + }, + }, + ], + } + } + + #[test] + fn aggregate_outcomes_keeps_per_link_codes() { + let outcomes = aggregate_outcomes(mixed_aggregate(ResponseCode::DoesNotExist)); + + assert_eq!(outcomes.len(), 2, "one outcome per link, in response order"); + assert_eq!(outcomes[0].0, LinkId::new("ok-link")); + assert!(outcomes[0].1.is_ok()); + assert_eq!(outcomes[1].0, LinkId::new("bad-link")); + // A failed link keeps the machine-readable code — the whole point of + // reporting per-node move outcomes instead of one formatted message. + match outcomes[1].1.as_ref().expect_err("bad-link must fail") { + ProtonError::Api(e) => { + assert_eq!(e.code, ResponseCode::DoesNotExist); + assert_eq!(e.message, "nope"); + assert_eq!(e.http_status, 200, "the batch request itself succeeded"); + } + other => panic!("expected an API error, got {other:?}"), + } + } + + #[test] + fn check_aggregate_reports_only_the_failed_links() { + check_aggregate("move", mixed_aggregate(ResponseCode::Success)) + .expect("all-success envelope must pass"); + + let err = check_aggregate("move", mixed_aggregate(ResponseCode::DoesNotExist)) + .expect_err("a failed link must fail the batch"); + let message = err.to_string(); + assert!(message.contains("move failed for 1 link(s)"), "{message}"); + assert!(message.contains("bad-link"), "{message}"); + assert!(!message.contains("ok-link"), "{message}"); + } + #[test] fn only_sealed_revisions_are_listable() { assert!(is_listable_revision_state(Some(1)), "active"); @@ -5856,4 +6720,128 @@ mod tests { assert_eq!(attrs.main_photo_link_id, None); assert!(attrs.tags.is_empty()); } + + fn digest(byte: u8) -> [u8; 32] { + [byte; 32] + } + + /// The pipeline stores blocks out of order; the manifest must not be. + #[test] + fn manifest_is_thumbnails_then_blocks_in_index_order() { + let manifest = assemble_manifest( + &[digest(0xaa), digest(0xbb)], + vec![(3, digest(3)), (1, digest(1)), (2, digest(2))], + ); + + assert_eq!(manifest.len(), 5 * 32); + let entries: Vec = manifest.chunks(32).map(|entry| entry[0]).collect(); + assert_eq!(entries, vec![0xaa, 0xbb, 1, 2, 3]); + } + + #[test] + fn manifest_of_a_thumbnail_less_empty_file_is_empty() { + assert!(assemble_manifest(&[], Vec::new()).is_empty()); + } + + fn target(index: Option, thumbnail_type: Option, token: &str) -> BlockUploadTarget { + BlockUploadTarget { + bare_url: "https://storage.example/blob".to_string(), + token: token.to_string(), + index, + thumbnail_type, + } + } + + #[test] + fn upload_targets_are_matched_by_index_not_position() { + let mut targets = vec![ + target(Some(2), None, "second"), + target(Some(1), None, "first"), + ]; + + assert_eq!( + take_block_target(&mut targets, 1) + .expect("target for 1") + .token, + "first" + ); + assert_eq!( + take_block_target(&mut targets, 2) + .expect("target for 2") + .token, + "second" + ); + assert!(take_block_target(&mut targets, 3).is_none()); + } + + /// A response that does not echo indices is consumed in request order. + #[test] + fn upload_targets_without_indices_fall_back_to_request_order() { + let mut targets = vec![target(None, None, "first"), target(None, None, "second")]; + + assert_eq!( + take_block_target(&mut targets, 7) + .expect("first target") + .token, + "first" + ); + assert_eq!( + take_block_target(&mut targets, 9) + .expect("second target") + .token, + "second" + ); + } + + #[test] + fn thumbnail_targets_are_matched_by_type() { + let mut targets = vec![ + target(None, Some(2), "preview"), + target(None, Some(1), "tiny"), + ]; + + assert_eq!( + take_thumbnail_target(&mut targets, 1) + .expect("target for type 1") + .token, + "tiny" + ); + assert_eq!( + take_thumbnail_target(&mut targets, 2) + .expect("target for type 2") + .token, + "preview" + ); + } + + fn api_error(code: ResponseCode, http_status: u16) -> ProtonError { + ProtonError::Api(ProtonApiError { + code, + http_status, + message: String::new(), + details: None, + }) + } + + #[test] + fn expired_upload_tokens_are_recognized_by_status_or_code() { + assert!(is_expired_upload_token(&api_error( + ResponseCode::Success, + 404 + ))); + assert!(is_expired_upload_token(&api_error( + ResponseCode::DoesNotExist, + 422 + ))); + + // Anything else is a plain failure: retried, but against the same token. + assert!(!is_expired_upload_token(&api_error( + ResponseCode::Success, + 500 + ))); + assert!(!is_expired_upload_token(&ProtonError::invalid_operation( + "boom" + ))); + assert!(!is_upload_timeout(&api_error(ResponseCode::Success, 504))); + } } diff --git a/crates/proton-drive-rs/src/crypto.rs b/crates/proton-drive-rs/src/crypto.rs index faf9368..f93c763 100644 --- a/crates/proton-drive-rs/src/crypto.rs +++ b/crates/proton-drive-rs/src/crypto.rs @@ -108,10 +108,31 @@ pub async fn decrypt_link_verified( link: &LinkDto, ) -> Result<(DecryptedLink, NodeVerification)> { let (passphrase, name, verification) = decrypt_link_parts(account, parent_key, link).await?; - let node_key = PrivateKey::from_armored(&link.key, &passphrase)?; + + // Unlocking the node key runs Proton's deliberately expensive S2K — tens of + // milliseconds of pure CPU, per node. Off the runtime it goes, so a folder + // listing decrypting many nodes at once uses more than one core and never + // stalls the reactor (the download path offloads its block crypto the same + // way, see `revision::decrypt_block_blocking`). + let armored_key = link.key.clone(); + let node_key = + blocking(move || Ok(PrivateKey::from_armored(&armored_key, &passphrase)?)).await?; + Ok((DecryptedLink { node_key, name }, verification)) } +/// Run CPU-bound PGP work on the blocking pool, mapping a join failure into a +/// [`ProtonError`]. +async fn blocking(work: F) -> Result +where + F: FnOnce() -> Result + Send + 'static, + T: Send + 'static, +{ + tokio::task::spawn_blocking(work) + .await + .map_err(|e| ProtonError::invalid_operation(format!("link decrypt task failed: {e}")))? +} + /// As [`decrypt_link_verified`], but stops short of unlocking the node's own /// key: it yields only what the parent key can read (the name, and the /// signature statuses). @@ -146,28 +167,40 @@ async fn decrypt_link_parts( AuthorshipClaim::create(account, link.name_signature_email.as_deref()).await }; - // Passphrase: decrypted with the parent key, signed (detached) by the node - // author; verify against the node claim (fallback = parent key). - let passphrase = parent_key.decrypt_armored_message(&link.passphrase)?; - let passphrase_status = match &link.passphrase_signature { - Some(sig) => verify_detached(sig, &passphrase, &node_claim.ring(parent_key)), - None => VerificationStatus::NotSigned, - }; + // The rings are resolved here (they may hit the key-resolution endpoint); + // everything after this point is pure PGP work and runs off the runtime. + let node_ring = node_claim.ring(parent_key); + let name_ring = name_claim.ring(parent_key); + let parent_key = parent_key.clone(); + let armored_passphrase = link.passphrase.clone(); + let passphrase_signature = link.passphrase_signature.clone(); + let armored_name = link.name.clone(); - // Name: an inline-signed message addressed to the parent key. - let (name_bytes, name_status) = - parent_key.decrypt_armored_verify(&link.name, &name_claim.ring(parent_key))?; - let name = String::from_utf8(name_bytes).map_err(|e| { - ProtonError::invalid_operation(format!("node name is not valid UTF-8: {e}")) - })?; + blocking(move || { + // Passphrase: decrypted with the parent key, signed (detached) by the + // node author; verify against the node claim (fallback = parent key). + let passphrase = parent_key.decrypt_armored_message(&armored_passphrase)?; + let passphrase_status = match &passphrase_signature { + Some(sig) => verify_detached(sig, &passphrase, &node_ring), + None => VerificationStatus::NotSigned, + }; - let verification = NodeVerification { - name: name_status, - passphrase: passphrase_status, - content_key: None, - extended_attributes: None, - }; - Ok((passphrase, name, verification)) + // Name: an inline-signed message addressed to the parent key. + let (name_bytes, name_status) = + parent_key.decrypt_armored_verify(&armored_name, &name_ring)?; + let name = String::from_utf8(name_bytes).map_err(|e| { + ProtonError::invalid_operation(format!("node name is not valid UTF-8: {e}")) + })?; + + let verification = NodeVerification { + name: name_status, + passphrase: passphrase_status, + content_key: None, + extended_attributes: None, + }; + Ok((passphrase, name, verification)) + }) + .await } /// Decrypt a file's content key and verify its `ContentKeyPacketSignature`. diff --git a/crates/proton-drive-rs/src/dtos.rs b/crates/proton-drive-rs/src/dtos.rs index 13945c8..782256d 100644 --- a/crates/proton-drive-rs/src/dtos.rs +++ b/crates/proton-drive-rs/src/dtos.rs @@ -174,7 +174,7 @@ impl LinkDetailsDto { } } -#[derive(Debug, Deserialize)] +#[derive(Debug, Clone, Deserialize)] pub struct LinkDto { #[serde(rename = "LinkID")] pub id: LinkId, @@ -837,7 +837,7 @@ pub struct BlockUploadPreparationRequest { pub thumbnails: Vec, } -#[derive(Debug, Serialize)] +#[derive(Debug, Clone, Serialize)] pub struct BlockCreationRequest { #[serde(rename = "Index")] pub index: i32, @@ -854,7 +854,7 @@ pub struct BlockCreationRequest { pub verifier: BlockVerifier, } -#[derive(Debug, Serialize)] +#[derive(Debug, Clone, Serialize)] pub struct BlockVerifier { /// Base64 verification token (`code XOR ciphertext_prefix`). #[serde(rename = "Token")] @@ -863,7 +863,7 @@ pub struct BlockVerifier { /// Thumbnail creation entry in a block-upload preparation request. Mirrors C# /// `ThumbnailCreationRequest` (`Size`, `Type`, base64 ciphertext `Hash`). -#[derive(Debug, Serialize)] +#[derive(Debug, Clone, Serialize)] pub struct ThumbnailCreationRequest { #[serde(rename = "Size")] pub size: i32, @@ -888,6 +888,16 @@ pub struct BlockUploadTarget { pub bare_url: String, #[serde(rename = "Token")] pub token: String, + /// Index of the content block this target is for, echoed from the request. + /// `Option` because it is absent on thumbnail targets and older responses; + /// the uploader then falls back to request order. TS reads the same field + /// (`UploadLinks[].Index` in `apiService.requestBlockUpload`). + #[serde(rename = "Index", default)] + pub index: Option, + /// Type of the thumbnail this target is for, echoed from the request + /// (TS `ThumbnailLinks[].ThumbnailType`). Absent on content-block targets. + #[serde(rename = "ThumbnailType", default)] + pub thumbnail_type: Option, } /// `PUT v2/volumes/{vid}/files/{lid}/revisions/{rid}` — seal the revision. @@ -1076,6 +1086,12 @@ pub struct SharedByMeLinkDto { pub share_id: ShareId, #[serde(rename = "LinkID")] pub link_id: LinkId, + /// The share the link is *reached through* — the context in which it lives, + /// which differs from `share_id` for an item inside a shared folder. C# + /// `SharedByMeLinkDto.ContextShareId`; optional here because the field is a + /// recent upstream addition and an older backend simply omits it. + #[serde(rename = "ContextShareID", default)] + pub context_share_id: Option, } /// The kind of item a share points at. C# `ShareTargetType`. diff --git a/crates/proton-drive-rs/src/lib.rs b/crates/proton-drive-rs/src/lib.rs index 9d56833..90d9050 100644 --- a/crates/proton-drive-rs/src/lib.rs +++ b/crates/proton-drive-rs/src/lib.rs @@ -44,6 +44,7 @@ mod photos; mod public_link; mod revision; mod sharing; +mod single_flight; pub use cache::{CachedNodeInfo, DriveEntityCache}; pub use client::ProtonDriveClient; diff --git a/crates/proton-drive-rs/src/photos.rs b/crates/proton-drive-rs/src/photos.rs index d3e18ea..121c5f4 100644 --- a/crates/proton-drive-rs/src/photos.rs +++ b/crates/proton-drive-rs/src/photos.rs @@ -112,6 +112,15 @@ impl ProtonPhotosClient { self.drive.enumerate_photos_timeline().await } + /// The photos I have shared with others, as [`NodeUid`]s — the photos-volume + /// counterpart of `ProtonDriveClient::enumerate_shared_by_me_node_uids`. + /// C# `ProtonPhotosClient.EnumerateSharedNodeUidsAsync`. Empty when the + /// account has no photos volume. Materialize with + /// [`enumerate_nodes`](Self::enumerate_nodes). + pub async fn enumerate_shared_node_uids(&self) -> Result> { + self.drive.enumerate_photos_shared_by_me_node_uids().await + } + /// Download and decrypt a photo's active revision, returning its plaintext. /// C# `PhotosFileDownloader`. pub async fn download_photo(&self, uid: &NodeUid) -> Result> { diff --git a/crates/proton-drive-rs/src/single_flight.rs b/crates/proton-drive-rs/src/single_flight.rs new file mode 100644 index 0000000..ad7800d --- /dev/null +++ b/crates/proton-drive-rs/src/single_flight.rs @@ -0,0 +1,158 @@ +//! Collapse concurrent identical loads into one request. +//! +//! Mirrors the intent of the TypeScript SDK's `internal/nodes/debouncer.ts` and +//! `internal/batchLoading.ts`: several callers asking for the same node — or the +//! same folder's key — at the same moment should cost one round-trip and one +//! decrypt, not one each. The batching half of those modules has no counterpart +//! here; this crate's loads are already issued in batches. + +use std::collections::HashMap; +use std::future::Future; +use std::hash::Hash; +use std::sync::Arc; + +use futures::future::{BoxFuture, FutureExt, Shared}; +use proton_sdk::error::{ProtonError, Result}; +use tokio::sync::Mutex; + +type SharedLoad = Shared>>>; + +/// A set of in-flight loads keyed by what they load. +pub(crate) struct SingleFlight { + inflight: Mutex>>, +} + +impl Default for SingleFlight { + fn default() -> Self { + Self { + inflight: Mutex::new(HashMap::new()), + } + } +} + +impl SingleFlight +where + K: Clone + Eq + Hash, + V: Clone + Send + 'static, +{ + /// Run `load` for `key`, or join the one already running for it. + /// + /// `load` must be self-contained (`'static`) because the caller that starts + /// it may go away before it finishes, leaving another to drive it. + /// + /// **Errors are shared, and sharing costs type fidelity.** [`ProtonError`] + /// is not `Clone` (its `Transport` variant wraps a `reqwest::Error`), so a + /// failed load hands every waiter — the one that started it included — a + /// rebuilt error: an API failure keeps its full [`ProtonApiError`], and + /// everything else degrades to + /// [`ProtonError::InvalidOperation`](proton_sdk::error::ProtonError::InvalidOperation) + /// carrying the original message. Callers that match on `ResponseCode` are + /// therefore unaffected; callers that match on the *variant* of a transport + /// failure would be, and none do. + pub(crate) async fn run(&self, key: K, load: F) -> Result + where + F: Future> + Send + 'static, + { + let shared = { + let mut inflight = self.inflight.lock().await; + match inflight.get(&key) { + Some(existing) => existing.clone(), + None => { + let load = load + .map(|outcome| outcome.map_err(Arc::new)) + .boxed() + .shared(); + inflight.insert(key.clone(), load.clone()); + load + } + } + }; + + let outcome = shared.await; + + // Clear the slot so the *next* caller gets a fresh load rather than this + // one's result. A racing arrival that already cloned the handle still + // gets this result, which is the point. + self.inflight.lock().await.remove(&key); + + outcome.map_err(|error| match error.as_ref() { + ProtonError::Api(e) => ProtonError::Api(e.clone()), + other => ProtonError::invalid_operation(other.to_string()), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use proton_sdk::api::ResponseCode; + use proton_sdk::error::ProtonApiError; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::sync::Notify; + + #[tokio::test] + async fn concurrent_loads_of_the_same_key_run_once() { + let flight: SingleFlight<&'static str, usize> = SingleFlight::default(); + let runs = Arc::new(AtomicUsize::new(0)); + + // The gate holds the load open until every waiter has registered — the + // case single-flighting exists for. A `Notify` rather than a one-shot so + // that a *failure* to deduplicate shows up as a wrong count, not a hang. + let gate = Arc::new(Notify::new()); + let waiters = (0..8).map(|_| { + let runs = runs.clone(); + let gate = gate.clone(); + flight.run("node", async move { + gate.notified().await; + Ok(runs.fetch_add(1, Ordering::SeqCst)) + }) + }); + + let (outcomes, ()) = tokio::join!(futures::future::join_all(waiters), async { + for _ in 0..4 { + tokio::task::yield_now().await; + } + gate.notify_waiters(); + }); + + for outcome in outcomes { + assert_eq!(outcome.expect("load"), 0); + } + assert_eq!(runs.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn a_later_load_is_not_served_the_previous_result() { + let flight: SingleFlight<&'static str, usize> = SingleFlight::default(); + assert_eq!(flight.run("node", async { Ok(1) }).await.expect("first"), 1); + assert_eq!( + flight.run("node", async { Ok(2) }).await.expect("second"), + 2 + ); + } + + #[tokio::test] + async fn a_shared_api_failure_keeps_its_response_code() { + let flight: SingleFlight<&'static str, usize> = SingleFlight::default(); + let error = flight + .run("node", async { + Err(ProtonError::Api(ProtonApiError { + code: ResponseCode::DoesNotExist, + http_status: 422, + message: "gone".into(), + details: None, + })) + }) + .await + .expect_err("load fails"); + + match error { + ProtonError::Api(e) => { + assert!(matches!(e.code, ResponseCode::DoesNotExist)); + assert_eq!(e.http_status, 422); + assert_eq!(e.message, "gone"); + } + other => panic!("expected an api error, got {other}"), + } + } +} diff --git a/crates/proton-drive-rs/tests/live_drive_ops.rs b/crates/proton-drive-rs/tests/live_drive_ops.rs index 1e73984..8dcc102 100644 --- a/crates/proton-drive-rs/tests/live_drive_ops.rs +++ b/crates/proton-drive-rs/tests/live_drive_ops.rs @@ -11,7 +11,7 @@ mod common; use proton_drive_rs::proton_sdk::crypto::VerificationStatus; use proton_drive_rs::{Node, NodeKind}; -use proton_sdk::ids::NodeUid; +use proton_sdk::ids::{LinkId, NodeUid}; /// Trash then permanently delete the given nodes; best-effort, logs on failure. async fn cleanup(client: &proton_drive_rs::ProtonDriveClient, uids: &[NodeUid]) { @@ -224,11 +224,18 @@ async fn move_nodes_batch() { .await .expect("create c2"); - client + let outcomes = client .move_nodes(&[c1.clone(), c2.clone()], &dst) .await .expect("move_nodes"); + assert_eq!(outcomes.len(), 2, "one outcome per requested node"); + for (uid, outcome) in &outcomes { + outcome + .as_ref() + .unwrap_or_else(|e| panic!("move of {uid} must succeed: {e}")); + } + assert_eq!( get(client, &c1, "c1 after move").await.parent_uid.as_ref(), Some(&dst), @@ -243,6 +250,66 @@ async fn move_nodes_batch() { cleanup(client, &[c1, c2, dst]).await; } +/// One bad node must not sink the batch: `move_nodes` reports per-node outcomes +/// (C# `MoveNodesAsync`'s result map), so a bogus link id fails alone while its +/// neighbour still moves. +#[tokio::test] +#[ignore = "live: needs test-account credentials"] +async fn move_nodes_partial_failure() { + let Some(live) = common::live_client().await else { + return; + }; + let client = &live.client; + + let root = client + .get_my_files_folder() + .await + .expect("get my-files root"); + let suffix = common::unique_suffix(); + + let dst = client + .create_folder(&root.uid, &format!("partial-dst-{suffix}"), None) + .await + .expect("create dst"); + let good = client + .create_folder(&root.uid, &format!("partial-good-{suffix}"), None) + .await + .expect("create good"); + let bogus = NodeUid::new( + dst.volume_id.clone(), + LinkId::new(format!("nonexistent-link-{suffix}")), + ); + + let outcomes = client + .move_nodes(&[good.clone(), bogus.clone()], &dst) + .await + .expect("move_nodes must not fail as a whole"); + + assert_eq!(outcomes.len(), 2, "one outcome per requested node"); + assert_eq!(outcomes[0].0, good, "outcomes keep the input order"); + outcomes[0] + .1 + .as_ref() + .expect("the real node must still move"); + assert_eq!(outcomes[1].0, bogus); + let e = outcomes[1] + .1 + .as_ref() + .expect_err("the bogus node must report its own failure"); + eprintln!("[info] bogus node outcome: {e}"); + + assert_eq!( + get(client, &good, "good after move") + .await + .parent_uid + .as_ref(), + Some(&dst), + "the real node must be under dst" + ); + + cleanup(client, &[good, dst]).await; +} + // --------------------------------------------------------------------------- // Signature verification // --------------------------------------------------------------------------- diff --git a/crates/proton-drive-rs/tests/live_sharing.rs b/crates/proton-drive-rs/tests/live_sharing.rs index ce5b3eb..b8f4435 100644 --- a/crates/proton-drive-rs/tests/live_sharing.rs +++ b/crates/proton-drive-rs/tests/live_sharing.rs @@ -18,7 +18,9 @@ mod common; -use proton_drive_rs::{MemberRole, ProtonDriveClient, ProtonDrivePublicLinkClient}; +use proton_drive_rs::{ + MemberRole, ProtonDriveClient, ProtonDrivePublicLinkClient, ProtonPhotosClient, +}; use proton_sdk::ids::NodeUid; /// Trash then permanently delete the given node; best-effort, logs on failure. @@ -251,6 +253,35 @@ async fn shared_with_me_and_incoming_invitations_read() { eprintln!("[info] incoming invitations count: {}", incoming.len()); } +/// The shared-by-me read surface must resolve on both volumes: the main volume +/// (`ProtonDriveClient`) and the photos volume (`ProtonPhotosClient`, C# +/// `EnumerateSharedNodeUidsAsync`). Empty is a pass — an account with nothing +/// shared, or with no photos volume at all, must not error. +#[tokio::test] +#[ignore = "live: needs test-account credentials"] +async fn shared_by_me_read() { + let Some(live) = common::live_client().await else { + return; + }; + let client = &live.client; + + let shared = client + .enumerate_shared_by_me_node_uids() + .await + .expect("enumerate_shared_by_me_node_uids must not error"); + eprintln!("[info] shared-by-me count (main volume): {}", shared.len()); + + let photos = ProtonPhotosClient::from_drive_client(client.clone()); + let shared_photos = photos + .enumerate_shared_node_uids() + .await + .expect("photos enumerate_shared_node_uids must not error"); + eprintln!( + "[info] shared-by-me count (photos volume): {}", + shared_photos.len() + ); +} + // --------------------------------------------------------------------------- // Public link — consuming a link as an anonymous visitor // --------------------------------------------------------------------------- diff --git a/crates/proton-drive-rs/tests/live_upload_download.rs b/crates/proton-drive-rs/tests/live_upload_download.rs index 2cfc0a4..76cd4f6 100644 --- a/crates/proton-drive-rs/tests/live_upload_download.rs +++ b/crates/proton-drive-rs/tests/live_upload_download.rs @@ -594,3 +594,135 @@ async fn revision_history_lists_reads_restores_and_deletes() { cleanup(client, &[file]).await; } + +/// Multi-block upload carrying thumbnails: their tokens now ride the first +/// block-token request instead of being prepared and stored one request each, +/// so the manifest layout (thumbnail digests first, in type order) has to +/// survive a pipeline that stores blocks out of order. +#[tokio::test] +#[ignore = "live: needs test-account credentials"] +async fn upload_download_multiblock_with_thumbnails_roundtrip() { + use proton_drive_rs::{Thumbnail, ThumbnailType}; + + let Some(live) = common::live_client().await else { + return; + }; + let client = &live.client; + + let root = client + .get_my_files_folder() + .await + .expect("get my-files root"); + + // 10 MiB → 3 content blocks, plus both thumbnail types. + let size = 10 * 1024 * 1024; + let mut payload = vec![0u8; size]; + for (i, b) in payload.iter_mut().enumerate() { + *b = (i as u32).wrapping_mul(2_654_435_761) as u8; + } + let small = b"small-thumbnail-bytes".to_vec(); + let preview = b"preview-thumbnail-bytes-which-are-longer".to_vec(); + + let name = format!("rt-thumbs-{}.bin", common::unique_suffix()); + let uid = client + .upload_file_from( + &root.uid, + &name, + "application/octet-stream", + std::io::Cursor::new(payload.clone()), + size as i64, + vec![ + // Deliberately out of type order: the uploader sorts. + Thumbnail::new(ThumbnailType::Preview, preview.clone()).expect("preview thumbnail"), + Thumbnail::new(ThumbnailType::Thumbnail, small.clone()).expect("small thumbnail"), + ], + None, // last_modification_time + false, // aead + ) + .await + .expect("upload_file_from with thumbnails"); + + assert_eq!( + client.download_file(&uid).await.expect("download_file"), + payload, + "content must round-trip alongside thumbnails" + ); + assert_eq!( + client + .download_thumbnail(&uid, ThumbnailType::Thumbnail) + .await + .expect("download small thumbnail"), + Some(small), + ); + assert_eq!( + client + .download_thumbnail(&uid, ThumbnailType::Preview) + .await + .expect("download preview thumbnail"), + Some(preview), + ); + + cleanup(client, &[uid]).await; +} + +/// Upload throughput, pipelined vs. serial — the measurement behind PERF_PLAN's +/// upload item, and the same methodology its download counterpart used. +/// +/// `with_max_inflight_blocks(1)` reproduces the pre-pipeline behavior exactly: +/// one block is read, encrypted and stored before the next one starts. The +/// default client keeps several blocks in flight. Both figures are printed +/// rather than asserted on — this is a benchmark, and a shared link's capacity +/// is not something a test can pin down. +#[tokio::test] +#[ignore = "live: benchmark, needs test-account credentials"] +async fn upload_throughput_pipelined_vs_serial() { + let Some(live) = common::live_client().await else { + return; + }; + let client = &live.client; + + let root = client + .get_my_files_folder() + .await + .expect("get my-files root"); + + // 40 MiB → 10 blocks, matching the download measurement's payload. + let size = 40 * 1024 * 1024; + let mut payload = vec![0u8; size]; + for (i, b) in payload.iter_mut().enumerate() { + *b = (i as u32).wrapping_mul(2_654_435_761) as u8; + } + + let serial = live.client.clone().with_max_inflight_blocks(1); + let mut uploaded = Vec::new(); + + for round in 1..=3 { + for (label, client) in [("serial", &serial), ("pipelined", client)] { + let name = format!("bench-{label}-{round}-{}.bin", common::unique_suffix()); + let started = std::time::Instant::now(); + let uid = client + .upload_file_from( + &root.uid, + &name, + "application/octet-stream", + std::io::Cursor::new(payload.clone()), + size as i64, + Vec::new(), + None, + false, + ) + .await + .expect("upload_file_from"); + let elapsed = started.elapsed(); + + eprintln!( + "[upload {label} round {round}] {:.1}s, {:.2} MiB/s", + elapsed.as_secs_f64(), + (size as f64 / (1024.0 * 1024.0)) / elapsed.as_secs_f64() + ); + uploaded.push(uid); + } + } + + cleanup(client, &uploaded).await; +} diff --git a/crates/proton-drive-rs/tests/spawnable.rs b/crates/proton-drive-rs/tests/spawnable.rs index b30b3da..e6027f6 100644 --- a/crates/proton-drive-rs/tests/spawnable.rs +++ b/crates/proton-drive-rs/tests/spawnable.rs @@ -1,12 +1,12 @@ -//! Compile-time guard: the download futures must survive `tokio::spawn`. +//! Compile-time guard: the transfer futures must survive `tokio::spawn`. //! -//! `download_file`/`download_file_to` and `RevisionReader::read_at` internally -//! drive a `futures::stream::buffered` pipeline. If any stage of that pipeline -//! borrows from a closure argument, the resulting future picks up a -//! higher-ranked lifetime and callers get "implementation of `FnOnce` is not -//! general enough" the moment they `tokio::spawn` it. That error surfaces only -//! *downstream* — this crate builds fine either way — so it went unnoticed until -//! a consumer broke. +//! `download_file`/`download_file_to`, `RevisionReader::read_at`, the upload +//! path and `enumerate_nodes` all internally drive a `futures::stream::buffered` +//! pipeline. If any stage of that pipeline borrows from a closure argument, the +//! resulting future picks up a higher-ranked lifetime and callers get +//! "implementation of `FnOnce` is not general enough" the moment they +//! `tokio::spawn` it. That error surfaces only *downstream* — this crate builds +//! fine either way — so it went unnoticed until a consumer broke. //! //! These never run; they only have to compile. @@ -46,6 +46,69 @@ fn download_range_is_spawnable( tokio::spawn(async move { client.download_range(&uid, 0, 4096).await }) } +/// `upload_file` inside a spawned task: the whole encrypt/store pipeline, from +/// an in-memory plaintext. +fn upload_file_is_spawnable( + client: ProtonDriveClient, + parent: NodeUid, +) -> JoinHandle> { + tokio::spawn(async move { + client + .upload_file( + &parent, + "spawned.bin", + "application/octet-stream", + &[0u8; 8], + ) + .await + }) +} + +/// `upload_file_from` inside a spawned task, streaming from an owned reader — +/// the shape the FUSE daemon's drain uses. +fn upload_file_from_is_spawnable( + client: ProtonDriveClient, + parent: NodeUid, +) -> JoinHandle> { + tokio::spawn(async move { + let plaintext: Vec = Vec::new(); + client + .upload_file_from( + &parent, + "spawned.bin", + "application/octet-stream", + std::io::Cursor::new(plaintext), + 0, + Vec::new(), + None, + false, + ) + .await + }) +} + +/// `upload_new_revision_from` inside a spawned task. +fn upload_new_revision_from_is_spawnable( + client: ProtonDriveClient, + uid: NodeUid, +) -> JoinHandle> { + tokio::spawn(async move { + let plaintext: Vec = Vec::new(); + client + .upload_new_revision_from(&uid, std::io::Cursor::new(plaintext), 0, Vec::new(), None) + .await + }) +} + +/// `enumerate_nodes` inside a spawned task: its per-node builds run through the +/// same `buffered` machinery. +fn enumerate_nodes_is_spawnable( + client: ProtonDriveClient, + uids: Vec, +) -> JoinHandle>> { + tokio::spawn(async move { client.enumerate_nodes(&uids).await }) +} + /// `open_revision` + a shared reader's `read_at` inside a spawned task — the /// shape an on-demand mount uses to fetch several blocks concurrently. fn reader_read_at_is_spawnable( diff --git a/crates/proton-sdk/src/http.rs b/crates/proton-sdk/src/http.rs index a391b0b..5d82833 100644 --- a/crates/proton-sdk/src/http.rs +++ b/crates/proton-sdk/src/http.rs @@ -206,16 +206,24 @@ impl ApiHttpClient { /// (filename `blob`, `application/octet-stream`) on the storage host, /// authorized by the per-block `pm-storage-token` header rather than the /// session bearer. The response is the usual JSON envelope. - pub async fn post_storage_blob(&self, url: &str, token: &str, blob: Vec) -> Result<()> { + /// + /// Takes [`Bytes`] rather than `Vec` because the multipart body has to be + /// rebuilt per attempt (a stream body can't be cloned) and a block is up to + /// 4 MiB: cloning `Bytes` bumps a refcount where cloning the `Vec` copied the + /// whole block on *every* attempt, first one included. The part is built with + /// an explicit length so the request still carries a `Content-Length`. + pub async fn post_storage_blob(&self, url: &str, token: &str, blob: Bytes) -> Result<()> { // Validate the part once up front; the multipart body itself is rebuilt - // per attempt inside the retry closure (a stream body can't be cloned). + // per attempt inside the retry closure. reqwest::multipart::Part::bytes(Vec::new()) .mime_str("application/octet-stream") .map_err(ProtonError::from)?; + let blob_len = blob.len() as u64; let mut timer = self.telemetry().start("storage_upload"); let response = send_retrying(&self.inner.config.retry_policy, || { - let part = reqwest::multipart::Part::bytes(blob.clone()) + let body = reqwest::Body::from(blob.clone()); + let part = reqwest::multipart::Part::stream_with_length(body, blob_len) .file_name("blob") .mime_str("application/octet-stream") .expect("octet-stream is a valid MIME type");