improved performance, concurrent upload, version 0.2.0

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",

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"

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. |

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

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<T, F>(work: F) -> Result<T>
+where
+    F: FnOnce() -> Result<T> + 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,17 +167,27 @@ 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)),
+    // 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();
+
+    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,
         };
 
         // 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))?;
+            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}"))
         })?;
@@ -168,6 +199,8 @@ async fn decrypt_link_parts(
             extended_attributes: None,
         };
         Ok((passphrase, name, verification))
+    })
+    .await
 }
 
 /// Decrypt a file's content key and verify its `ContentKeyPacketSignature`.

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<ThumbnailCreationRequest>,
 }
 
-#[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<i32>,
+    /// 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<i32>,
 }
 
 /// `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<ShareId>,
 }
 
 /// The kind of item a share points at. C# `ShareTargetType`.

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;

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<Vec<NodeUid>> {
+        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<Vec<u8>> {

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<V> = Shared<BoxFuture<'static, std::result::Result<V, Arc<ProtonError>>>>;
+
+/// A set of in-flight loads keyed by what they load.
+pub(crate) struct SingleFlight<K, V> {
+    inflight: Mutex<HashMap<K, SharedLoad<V>>>,
+}
+
+impl<K, V> Default for SingleFlight<K, V> {
+    fn default() -> Self {
+        Self {
+            inflight: Mutex::new(HashMap::new()),
+        }
+    }
+}
+
+impl<K, V> SingleFlight<K, V>
+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<F>(&self, key: K, load: F) -> Result<V>
+    where
+        F: Future<Output = Result<V>> + 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}"),
+        }
+    }
+}

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
 // ---------------------------------------------------------------------------

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
 // ---------------------------------------------------------------------------

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;
+}

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<proton_sdk::error::Result<NodeUid>> {
+    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<proton_sdk::error::Result<NodeUid>> {
+    tokio::spawn(async move {
+        let plaintext: Vec<u8> = 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<proton_sdk::error::Result<()>> {
+    tokio::spawn(async move {
+        let plaintext: Vec<u8> = 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<NodeUid>,
+) -> JoinHandle<proton_sdk::error::Result<Vec<proton_drive_rs::Node>>> {
+    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(

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<u8>) -> Result<()> {
+    ///
+    /// Takes [`Bytes`] rather than `Vec<u8>` 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");