mirror of
https://github.com/narrrl/proton-sdk-rs.git
synced 2026-11-03 16:41:26 +01:00
feat(drive,sdk): port upstream C# SDK up to 6cbf2f44
Port 6cbf2f44 (stop retrying permanent API errors on block transfers) and 20822176 (small file upload interop); 6ce47922 is a no-op here. - Add ProtonApiError::is_retriable / ProtonError::is_retriable mirroring C# RetryPolicy.IsRetriable: 4xx is permanent except 404, 408 and 429, with envelope codes able to override the HTTP status. - Gate upload_one_block's retry loop on it so a permanent rejection of a revision fails immediately instead of being replayed three times. - Extract BlockTargets from RevisionReader so the whole-file download path, the public link download and the reader all re-mint expired block URLs through one generation-deduplicated refresh. - Fall back from the small upload to the block path on AlreadyExists only; the small upload is not idempotent, so everything else keeps propagating. - Bound a server-supplied Retry-After by RetryPolicy::max_retry_after (default 120 s) so a long rate-limit answer cannot look like a hang.
This commit is contained in:
nvrl committed
1 parent
4f129b778a
commit
1ca28c5848
7 files changed
+420
-113
No files matched your search
+3
-2
@@ -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**: `ce000edd053768bf8fd2da04f3ef4fa68ba9e71e`
|
||||
- **Date**: 2026-09-15
|
||||
- **Pinned**: `6cbf2f44`
|
||||
- **Date**: 2026-09-19
|
||||
|
||||
> **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.
|
||||
@@ -38,3 +38,4 @@ gitignored, so this file is the only durable record of what we last reviewed.
|
||||
| 2026-08-18 | — (no upstream delta) | — | **not an upstream port of C#** — ported from the **TypeScript** SDK, which is the only upstream with an album write API. Ported: **album creation** (`ProtonPhotosClient::create_album` — `POST photos/volumes/{vid}/albums`, from `AlbumsManager.createAlbum` + `AlbumsCryptoService.createAlbum`; an album is a folder node parented to the photos root, so the crypto is the folder create: generated node key, passphrase encrypted to the photos root key and signed by the photos share's membership address, plus the album's own child-name hash key, and TS's `Locked: false`) and **adding photos to an album** (`add_photos_to_album` — `POST photos/volumes/{vid}/albums/{lid}/add-multiple`, from `AddToAlbumProcess.processSameVolumeQueue` + `PhotoTransferPayloadBuilder`; per photo the name is re-encrypted to the album key, name/content hashes recomputed under the album hash key, and the node passphrase **rewrapped** via `rewrap_message_to` rather than re-encrypted from plaintext — the ciphertext and therefore the original `NodePassphraseSignature` survive, which is why the entry carries no signature of its own; related photos are flattened in beside their main photo, batches of 10 like TS's `ADD_PHOTOS_BATCH_SIZE`, one outcome per input photo in input order like `move_nodes`). Not ported: TS `copyPhoto` (adding a photo that lives on **another** volume, i.e. from an album shared with us — needs the cross-volume copy path; those inputs fail their own outcome), `updateAlbum` (rename / cover photo) and `deleteAlbum`. Also **ours, with no upstream counterpart**: `find_duplicates_many` and `name_collisions` — the duplicates endpoint already takes a `NameHashes` array but both SDKs only ever send one, and a Google Takeout import asks the question tens of thousands of times; `name_collisions` answers on the name hash alone so the caller only has to read and SHA-1 the files whose names actually collide. Live validation pending: everything here needs a photos-volume account. |
|
||||
| 2026-08-27 | `df6b6bfd` → `c8d03244` | `e0e198e2`, `c8d03244` | ported: **save photos to the timeline** (`ProtonPhotosClient::save_photos_to_timeline` — C# `PhotoOperations.SavePhotosToTimelineAsync`, itself the JS `saveToTimeline`). Photos already on our photos volume are *moved* into the timeline root through `PUT photos/volumes/{vid}/links/transfer-multiple`, batched so no request carries more than ten links (main photo + related photos counted together, never split across requests); photos on another volume — from a shared-with-me album — are *copied* through `POST volumes/{vid}/links/{lid}/copy`, one request each, with their related photos nested in `Photos`. This closes the **cross-volume copy** hole logged in the 2026-08-18 row (TS `copyPhoto`), so photos from albums shared with us can now be kept. The re-encryption is what `add_photos_to_album` already did — name re-encrypted to the target key and signed by the photos membership address, name/content hashes recomputed under the target hash key, node passphrase **rewrapped** via `rewrap_message_to` so the original `NodePassphraseSignature` survives — so it was lifted into a shared `photo_target_payload` (C# `PhotosCrypto.EncryptPhotoForTarget` + `PhotoTransferPayloadBuilder`) that both paths build on. Retry semantics mirror upstream exactly: a photo the server reports an incomplete group for is re-queued **once** with the related photos it named, and only its second answer becomes an outcome; a photo already parented to the timeline root fails its own outcome. Also ported: `ProtonPhotosClient::enumerate_events` (`c8d03244`'s only non-interop hunk — a passthrough to the Drive client's, which we already had). diverged deliberately: upstream returns `IAsyncEnumerable<NodeActionResult>`; ours collects `Vec<(NodeUid, Result<()>)>` like `move_nodes` and the other batch ops — and, because of the retry queue, outcome order follows completion, not input, which the API docs state. not ported: `29e58378` **per-file upload performance metrics** (`UploadPerformanceEvent`, size/route/block-count classification, active-vs-paused attempt timing on `UploadController`) — it is telemetry plumbing for C#'s interop metric callback; our `proton-sdk::telemetry` has its own `TelemetryEvent`/`OpTimer` shape and we have no pause/resume upload controller, and nothing about the upload itself changes. structural, not ported: `0537d906` (C# merges its batch loaders and `DtoToMetadataConverter` into one `NodeProvider`; the only behavior-adjacent bit is dropping the cache's claim/acquisition dance, which our single-flight already covers). Live validation pending: the whole path needs a photos-volume account, and the cross-volume half needs a second account sharing an album with it. |
|
||||
| 2026-09-15 | `c8d03244` → `ce000edd` | `cdc0f260`, `4b1129aa`, `63a7fc1e`, `5cbbb0ad`, `e33712b9` | ported: **trash time field name** (`cdc0f260` — C# dropped a `[JsonPropertyName("Trashed")]` that never matched the API's `TrashTime`; ours had copied the same wrong name, so `LinkDto` now reads `TrashTime` and keeps `Trashed` as a separate legacy field rather than a serde alias, which would fail a link carrying both). **direct role + membership detail** (`4b1129aa`): `Node::direct_role` — `Admin` on our own main or photos volume (`is_own_volume`, C# `VolumeOperations.IsOwnVolumeAsync`), otherwise the membership's permissions or `Inherited`; `ShareMembership` gains `invite_time`, `inviter_email` and `inviter_verification`, the last from checking the inviter's `drive.share-member.inviter` signature over our share-passphrase key packet against the claimed inviter's keys (C# `NodeCrypto.VerifyMembershipInviter`; new `proton_sdk::crypto::verify_detached_with_context`, which requires the `context@proton.ch` notation to match and accepts armored or base64-binary signatures). Non-fatal like every other verification. Kept optional (`Option<MemberRole>` / `Option<VerificationStatus>`, `skip_serializing_if` on the membership additions) so persisted `Node` blobs keep deserializing and read back as "unknown" rather than as a wrong role; C# models `SharedBy` as `Result<Author, SignatureVerificationError>` — ours is the claimed email plus the status. **share id** (`63a7fc1e`): `Node::share_id`, the `Sharing` block's `ShareID` (C# `DeprecatedShareId`, internal there; public here as the JS SDK has it). **batched move with rename, streamed** (`5cbbb0ad` + `e33712b9` — upstream finally replaced its per-node loop with the `move-multiple` endpoint we already used): new `move_nodes_streaming(Vec<NodeMoveItem>, NodeUid)` yielding `(NodeUid, Result<()>)` per node as each batch lands, with `NodeMoveItem::target_name` for rename-on-move; batches of 150 that never mix anonymously signed nodes with signed ones; an anonymously signed node now gets its passphrase signed by the mover plus `SignatureEmail` — **also fixed in `move_node`**, which never did this (C# `MoveSingleAsync` always had); a link rejected `InvalidRequirements` (2000) is retried once with the server's `NameHash` provided parent and decrypted name are unchanged, else fails out of sync (C# `PrepareItemWithRemoteNameDigestAsync`). `move_nodes` keeps its signature and input-order contract and now collects the stream. diverged deliberately: `NodeMoveItem` carries no `CurrentParentUid`/`CurrentName` — we read both off the link rather than trust the caller, so upstream's retry mostly exists for a case we only hit through a stale cached `OriginalHash`; per-code exception types (`NodeWithSameNameExistsException`, `FolderCapacityExceededException`, `FolderNestingTooDeepException`, `CannotMoveFavoritePhotoFromShareException`, `NodeOutOfSyncException`, `ParentFolderNotFoundException`) are not ported — the per-link `ProtonError::Api` keeps the `ResponseCode` to match on; `RenameNodeAsync` removal not mirrored (`rename_node` stays, `proton-drive-linux` uses it). no-op: `e3d602b9` (C# `HashingReadStream`/`HashingWriteStream` hashed the whole buffer instead of `offset..count`; our upload hashes `buf.truncate(n)`-ed blocks and has no stream wrapper). noise: `df87c3aa` (AEAD feature-flag removal, chore). **Downstream:** adding `Node::{direct_role, share_id}` and `ShareMembership::{invite_time, inviter_email, inviter_verification}` breaks struct literals in `proton-drive-linux` tests (`pdfs-fuse/src/state.rs`) at the next SDK bump. Live validation pending for all of it — membership verification and anonymous-node moves need a second account.|
|
||||
| 2026-09-19 | `ce000edd` → `6cbf2f44` | `6cbf2f44`, `20822176` | **retriability on block transfers** (`6cbf2f44`): new `ProtonError::is_retriable()` / `ProtonApiError::is_retriable()` (C# `RetryPolicy.IsRetriable` — a 4xx is permanent except 404, 408 and 429; the envelope codes `TooManyRequests` and `DoesNotExist` override the status, as a rate limit can arrive under a 2xx-shaped envelope). `upload_one_block` now ends the transfer on a permanent rejection instead of spending its remaining attempts on it, keeping the existing expired-upload-token re-mint and timeout downshift. The whole-file download path gained the expired-URL recovery `RevisionReader` already had: the block table plus its deduplicated one-shot re-list moved into `revision::BlockTargets`, which `RevisionReader::block_plaintext`, `ProtonDriveClient::write_content_blocks` and the public-link `download_file_to` all fetch through — so a download long enough to outlive its block URLs re-lists the revision once per expiry rather than failing. diverged deliberately: thumbnail blob fetches got no change — upstream passes them `refreshTransferTargetAsync: null` and relies on the same predicate to stop retrying, and our thumbnail fetch is a single `get_storage_blob` whose only retry is `http::send_retrying`, which already retries 408/429/502/503/504 and nothing else. **small-upload interop** (`20822176`): `upload_file` / `upload_new_revision` now fall back to the regular block path when the one-request endpoint reports `AlreadyExists` (2500) — the Proton API reports upstream's `NodeWithSameNameExists` and `RevisionDraftConflict` under that one code, and the block path knows how to delete our own stale draft and retry while the small endpoint is one shot. Everything else still propagates: a small upload creates, uploads and seals in a single non-idempotent request, so replaying it on 429/424/5xx/a dropped connection risks a duplicate file (upstream excludes 429 and 424 explicitly). Not ported: upstream's `SmallUploadNotApplicable` condition has no envelope code we can recognise, and `HttpRequestType.SmallUpload` maps to our telemetry op name rather than a new request type. no-op: `6ce47922` (thumbnail retry leaking upload-queue slots) — we have no `UploadQueue` and no per-file block accounting, only the client-wide in-flight block semaphore, and our thumbnail list is a slice rather than a lazily re-enumerated `IEnumerable`, so the double-reservation it fixes cannot occur. **Ours, not upstream**: `RetryPolicy::max_retry_after` (default 120 s) now bounds a server-supplied `Retry-After`, which was honoured uncapped inside a sleep a daemon thread blocks on — an hour-long one is indistinguishable from a hang. Live validation pending: a multi-block upload, a <128 KiB upload, and a download left running past its block-URL lifetime.|
|
||||
@@ -84,7 +84,8 @@ use crate::photos::{
|
||||
AlbumItem, PhotoTag, PhotoTagsUpdate, PhotoUploadMetadata, PhotosTimelineItem,
|
||||
};
|
||||
use crate::revision::{
|
||||
MAX_CONCURRENT_BLOCK_DOWNLOADS, Revision, RevisionReader, digest_and_decrypt_block_blocking,
|
||||
BlockTargets, MAX_CONCURRENT_BLOCK_DOWNLOADS, Revision, RevisionReader,
|
||||
digest_and_decrypt_block_blocking,
|
||||
};
|
||||
use crate::sharing::{
|
||||
Bookmark, ExternalInvitation, ExternalInvitationState, IncomingInvitation, MemberRole,
|
||||
@@ -1304,7 +1305,15 @@ impl ProtonDriveClient {
|
||||
}
|
||||
}
|
||||
|
||||
self.write_content_blocks(&blocks, &content_key, &mut manifest, output)
|
||||
let block_count = blocks.len();
|
||||
let targets = Arc::new(BlockTargets::new(
|
||||
self.revision_transport(),
|
||||
uid.clone(),
|
||||
revision_id.clone(),
|
||||
blocks,
|
||||
block_count,
|
||||
));
|
||||
self.write_content_blocks(targets, block_count, &content_key, &mut manifest, output)
|
||||
.await?;
|
||||
|
||||
verify_manifest(&self.account, &revision, &node_key, &manifest).await;
|
||||
@@ -1990,9 +1999,16 @@ impl ProtonDriveClient {
|
||||
// reader bounds or buffering arbitrary inputs.
|
||||
if self.small_file_upload.load(Ordering::Relaxed) && small_upload_applicable(contents.len())
|
||||
{
|
||||
return self
|
||||
match self
|
||||
.upload_small_file(parent_uid, name, media_type, contents)
|
||||
.await;
|
||||
.await
|
||||
{
|
||||
Ok(uid) => return Ok(uid),
|
||||
Err(error) if small_upload_should_fall_back(&error) => {
|
||||
tracing::debug!(%error, "small file upload conflicted; falling back to blocks");
|
||||
}
|
||||
Err(error) => return Err(error),
|
||||
}
|
||||
}
|
||||
self.upload_file_from(
|
||||
parent_uid,
|
||||
@@ -2217,7 +2233,13 @@ impl ProtonDriveClient {
|
||||
pub async fn upload_new_revision(&self, file_uid: &NodeUid, contents: &[u8]) -> Result<()> {
|
||||
if self.small_file_upload.load(Ordering::Relaxed) && small_upload_applicable(contents.len())
|
||||
{
|
||||
return self.upload_small_revision(file_uid, contents, None).await;
|
||||
match self.upload_small_revision(file_uid, contents, None).await {
|
||||
Ok(()) => return Ok(()),
|
||||
Err(error) if small_upload_should_fall_back(&error) => {
|
||||
tracing::debug!(%error, "small revision upload conflicted; falling back to blocks");
|
||||
}
|
||||
Err(error) => return Err(error),
|
||||
}
|
||||
}
|
||||
self.upload_new_revision_from(
|
||||
file_uid,
|
||||
@@ -5274,23 +5296,20 @@ impl ProtonDriveClient {
|
||||
/// `output` depend on that ordering.
|
||||
async fn write_content_blocks<W: std::io::Write>(
|
||||
&self,
|
||||
blocks: &[BlockDto],
|
||||
targets: Arc<BlockTargets>,
|
||||
block_count: usize,
|
||||
content_key: &ContentKey,
|
||||
manifest: &mut Vec<u8>,
|
||||
output: &mut W,
|
||||
) -> Result<()> {
|
||||
// Each fetch owns its url/token rather than borrowing `blocks`. Borrowing
|
||||
// here makes the resulting future carry a higher-ranked lifetime that
|
||||
// `tokio::spawn` rejects ("implementation of `FnOnce` is not general
|
||||
// enough") in *callers* of `download_file_to` — a downstream break, not a
|
||||
// local one, so it does not show up in this crate's own build.
|
||||
let fetches: Vec<(String, String)> = blocks
|
||||
.iter()
|
||||
.map(|block| (block.bare_url.clone(), block.token.clone()))
|
||||
.collect();
|
||||
|
||||
let mut decrypted = stream::iter(fetches.into_iter().map(|(url, token)| {
|
||||
let http = self.http.clone();
|
||||
// Each fetch owns an `Arc` of the block table rather than borrowing it.
|
||||
// Borrowing here makes the resulting future carry a higher-ranked
|
||||
// lifetime that `tokio::spawn` rejects ("implementation of `FnOnce` is
|
||||
// not general enough") in *callers* of `download_file_to` — a downstream
|
||||
// break, not a local one, so it does not show up in this crate's own
|
||||
// build.
|
||||
let mut decrypted = stream::iter((0..block_count).map(|index| {
|
||||
let targets = targets.clone();
|
||||
let content_key = content_key.clone();
|
||||
let slots = self.block_slots();
|
||||
async move {
|
||||
@@ -5301,7 +5320,10 @@ impl ProtonDriveClient {
|
||||
let permit = slots.acquire_owned().await.map_err(|e| {
|
||||
ProtonError::invalid_operation(format!("block slots closed: {e}"))
|
||||
})?;
|
||||
let ciphertext = http.get_storage_blob(&url, &token).await?;
|
||||
// Goes through the block table, not a snapshot of it: a download
|
||||
// of a large file outlives its block URLs, and this re-lists the
|
||||
// revision once per expiry instead of failing the transfer.
|
||||
let ciphertext = targets.ciphertext(index).await?;
|
||||
let (digest, plaintext) =
|
||||
digest_and_decrypt_block_blocking(content_key, ciphertext).await?;
|
||||
Ok::<_, ProtonError>((digest, plaintext, permit))
|
||||
@@ -6985,7 +7007,15 @@ impl ProtonDriveClient {
|
||||
}
|
||||
}
|
||||
|
||||
self.write_content_blocks(&blocks, &content_key, &mut manifest, output)
|
||||
let block_count = blocks.len();
|
||||
let targets = Arc::new(BlockTargets::new(
|
||||
self.revision_transport(),
|
||||
uid.clone(),
|
||||
revision_id.clone(),
|
||||
blocks,
|
||||
block_count,
|
||||
));
|
||||
self.write_content_blocks(targets, block_count, &content_key, &mut manifest, output)
|
||||
.await?;
|
||||
|
||||
verify_manifest(&self.account, &revision, &node_key, &manifest).await;
|
||||
@@ -7969,6 +7999,26 @@ fn small_upload_applicable(plaintext_size: usize) -> bool {
|
||||
plaintext_size.saturating_mul(11) < SMALL_UPLOAD_SIZE_LIMIT * 10
|
||||
}
|
||||
|
||||
/// Whether a failed one-request small upload should be retried on the regular
|
||||
/// block path.
|
||||
///
|
||||
/// Upstream (C# `SmallFileUploadFallback`) falls back on exactly three
|
||||
/// conditions: the endpoint declaring itself not applicable, a node of the same
|
||||
/// name already existing, and a revision-draft conflict. The Proton API reports
|
||||
/// the latter two as the same envelope code, `AlreadyExists` (2500), which the
|
||||
/// regular path then resolves — it knows how to delete a stale draft of ours and
|
||||
/// retry (`create_file_draft` / `create_revision_draft`), while the small
|
||||
/// endpoint is one shot and cannot.
|
||||
///
|
||||
/// Everything else propagates. A small upload is **not** idempotent: its single
|
||||
/// request creates the node, uploads the block and seals the revision together,
|
||||
/// so replaying it on a rate limit (429), a failed dependency (424), a 5xx or a
|
||||
/// dropped connection risks a duplicate file. Those are the caller's to retry,
|
||||
/// with the same knowledge of what already landed.
|
||||
fn small_upload_should_fall_back(error: &ProtonError) -> bool {
|
||||
matches!(error, ProtonError::Api(e) if e.code == ResponseCode::AlreadyExists)
|
||||
}
|
||||
|
||||
/// Block verification token: `verificationCode XOR ciphertextPrefix`, with the
|
||||
/// ciphertext prefix zero-padded or truncated to the code length. Mirrors C#
|
||||
/// `VerificationToken.Create`.
|
||||
@@ -8159,6 +8209,14 @@ async fn upload_one_block(
|
||||
if attempt >= MAX_BLOCK_UPLOAD_ATTEMPTS {
|
||||
return Err(error);
|
||||
}
|
||||
// A permanent rejection of the revision (C# `RetryPolicy.IsRetriable`)
|
||||
// fails the same way on every replay, so it ends the transfer here
|
||||
// rather than spending the remaining attempts on it. 404 stays
|
||||
// retriable: it is how an expired upload token arrives, and the branch
|
||||
// below re-mints one.
|
||||
if !error.is_retriable() {
|
||||
return Err(error);
|
||||
}
|
||||
|
||||
if is_upload_timeout(&error) {
|
||||
downshifted.store(true, Ordering::Relaxed);
|
||||
@@ -8372,8 +8430,8 @@ mod tests {
|
||||
context_share_path, drive_items, epoch_to_iso8601, is_expired_upload_token,
|
||||
is_listable_revision_state, is_upload_timeout, join_node_path, missing_related_links,
|
||||
path_segments, record_timeline_outcome, run_context_share_mutation,
|
||||
share_membership_from_dto, small_upload_applicable, take_block_target,
|
||||
take_thumbnail_target, to_drive_event, validate_node_name,
|
||||
share_membership_from_dto, small_upload_applicable, small_upload_should_fall_back,
|
||||
take_block_target, take_thumbnail_target, to_drive_event, validate_node_name,
|
||||
};
|
||||
use crate::dtos::{
|
||||
AggregateLinksResponse, BlockUploadTarget, LinkIdResponsePair, ShareMembershipSummaryDto,
|
||||
@@ -9115,4 +9173,35 @@ mod tests {
|
||||
};
|
||||
assert!(missing_related_links(&plain, &volume_id).is_empty());
|
||||
}
|
||||
|
||||
/// The one condition the regular block path can resolve and the one-request
|
||||
/// endpoint cannot: a name/draft conflict, which the block path clears by
|
||||
/// deleting our own stale draft.
|
||||
#[test]
|
||||
fn a_name_or_draft_conflict_falls_back_to_the_block_path() {
|
||||
assert!(small_upload_should_fall_back(&api_error(
|
||||
ResponseCode::AlreadyExists,
|
||||
422
|
||||
)));
|
||||
}
|
||||
|
||||
/// A small upload is not idempotent, so a transient failure must not be
|
||||
/// replayed by us — it could duplicate the file.
|
||||
#[test]
|
||||
fn transient_failures_do_not_fall_back() {
|
||||
for (status, code) in [
|
||||
(429, ResponseCode::TooManyRequests),
|
||||
(424, ResponseCode::Unknown),
|
||||
(503, ResponseCode::ServiceUnavailable),
|
||||
(422, ResponseCode::InsufficientQuota),
|
||||
] {
|
||||
assert!(
|
||||
!small_upload_should_fall_back(&api_error(code, status)),
|
||||
"http {status} should not fall back"
|
||||
);
|
||||
}
|
||||
assert!(!small_upload_should_fall_back(
|
||||
&ProtonError::invalid_operation("bug")
|
||||
));
|
||||
}
|
||||
}
|
||||
@@ -59,7 +59,9 @@ use crate::dtos::{
|
||||
ThumbnailBlockListRequest, ThumbnailBlockListResponse,
|
||||
};
|
||||
use crate::node::{FileThumbnail, Node, NodeKind, RevisionState, ThumbnailType};
|
||||
use crate::revision::{MAX_CONCURRENT_BLOCK_DOWNLOADS, RevisionReader, decrypt_block_blocking};
|
||||
use crate::revision::{
|
||||
BlockTargets, MAX_CONCURRENT_BLOCK_DOWNLOADS, RevisionReader, decrypt_block_blocking,
|
||||
};
|
||||
use crate::sharing::MemberRole;
|
||||
use crate::single_flight::SingleFlight;
|
||||
use crate::transport::{BlockSession, RevisionTransport, rank_block_sizes};
|
||||
@@ -635,17 +637,27 @@ impl ProtonDrivePublicLinkClient {
|
||||
// Fetched and decrypted several at a time but yielded in block order, so
|
||||
// the writer still sees the file sequentially and no more than
|
||||
// `MAX_CONCURRENT_BLOCK_DOWNLOADS` blocks are ever resident.
|
||||
let mut plaintexts = stream::iter(blocks.into_iter().map(|block| {
|
||||
let block_count = blocks.len();
|
||||
// Shared with the authenticated download path: block URLs expire on a
|
||||
// server-side schedule we are not told, so a long transfer re-lists the
|
||||
// revision once per expiry rather than failing.
|
||||
let targets = Arc::new(BlockTargets::new(
|
||||
self.transport.clone(),
|
||||
uid.clone(),
|
||||
revision_id,
|
||||
blocks,
|
||||
block_count,
|
||||
));
|
||||
|
||||
let mut plaintexts = stream::iter((0..block_count).map(|index| {
|
||||
let transport = self.transport.clone();
|
||||
let targets = targets.clone();
|
||||
let content_key = content_key.clone();
|
||||
async move {
|
||||
let permit = transport.block_slots().acquire_owned().await.map_err(|e| {
|
||||
ProtonError::invalid_operation(format!("block slots closed: {e}"))
|
||||
})?;
|
||||
let ciphertext = transport
|
||||
.http()
|
||||
.get_storage_blob(&block.bare_url, &block.token)
|
||||
.await?;
|
||||
let ciphertext = targets.ciphertext(index).await?;
|
||||
let plaintext = decrypt_block_blocking(content_key, ciphertext).await?;
|
||||
Ok::<_, ProtonError>((plaintext, permit))
|
||||
}
|
||||
|
||||
@@ -121,6 +121,125 @@ struct BlockTable {
|
||||
generation: u64,
|
||||
}
|
||||
|
||||
/// A revision's block table plus the ability to re-mint it.
|
||||
///
|
||||
/// Block URLs carry their own authorization and are handed out with a
|
||||
/// server-side lifetime we are never told, so any download long enough to
|
||||
/// outlive one has to re-list the revision mid-flight. That recovery is
|
||||
/// identical for a seekable [`RevisionReader`] and for a whole-file download, so
|
||||
/// it lives here and both go through it.
|
||||
pub(crate) struct BlockTargets {
|
||||
transport: RevisionTransport,
|
||||
uid: NodeUid,
|
||||
revision_id: String,
|
||||
blocks: RwLock<BlockTable>,
|
||||
/// Serializes block-table refreshes so a burst of expired-URL failures
|
||||
/// triggers one re-listing rather than one per block.
|
||||
refresh: Mutex<()>,
|
||||
/// How many blocks the caller resolved the revision as having. A re-list
|
||||
/// that disagrees means the revision changed underneath the transfer, which
|
||||
/// no amount of retrying fixes.
|
||||
expected_blocks: usize,
|
||||
}
|
||||
|
||||
impl BlockTargets {
|
||||
pub(crate) fn new(
|
||||
transport: RevisionTransport,
|
||||
uid: NodeUid,
|
||||
revision_id: String,
|
||||
blocks: Vec<BlockDto>,
|
||||
expected_blocks: usize,
|
||||
) -> Self {
|
||||
Self {
|
||||
transport,
|
||||
uid,
|
||||
revision_id,
|
||||
blocks: RwLock::new(BlockTable {
|
||||
blocks,
|
||||
generation: 0,
|
||||
}),
|
||||
refresh: Mutex::new(()),
|
||||
expected_blocks,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn uid(&self) -> &NodeUid {
|
||||
&self.uid
|
||||
}
|
||||
|
||||
pub(crate) fn revision_id(&self) -> &str {
|
||||
&self.revision_id
|
||||
}
|
||||
|
||||
/// The storage URL and token for a block, plus the generation of the table
|
||||
/// they came from.
|
||||
async fn location(&self, index: usize) -> Result<(String, String, u64)> {
|
||||
let table = self.blocks.read().await;
|
||||
let block = table.blocks.get(index).ok_or_else(|| {
|
||||
ProtonError::invalid_operation(format!(
|
||||
"block {index} is missing from revision {}",
|
||||
self.revision_id
|
||||
))
|
||||
})?;
|
||||
Ok((
|
||||
block.bare_url.clone(),
|
||||
block.token.clone(),
|
||||
table.generation,
|
||||
))
|
||||
}
|
||||
|
||||
/// Re-list the revision to obtain fresh block URLs, unless another task has
|
||||
/// already replaced the generation the caller saw.
|
||||
async fn refresh(&self, seen_generation: u64) -> Result<()> {
|
||||
let _guard = self.refresh.lock().await;
|
||||
|
||||
if self.blocks.read().await.generation != seen_generation {
|
||||
// Someone else refreshed while we waited; their table is at least as
|
||||
// fresh as one we would fetch now.
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let (_, blocks) = self
|
||||
.transport
|
||||
.list_blocks(&self.uid.volume_id, &self.uid.link_id, &self.revision_id)
|
||||
.await?;
|
||||
|
||||
if blocks.len() != self.expected_blocks {
|
||||
return Err(ProtonError::invalid_operation(format!(
|
||||
"revision {} changed block count while open ({} -> {})",
|
||||
self.revision_id,
|
||||
self.expected_blocks,
|
||||
blocks.len()
|
||||
)));
|
||||
}
|
||||
|
||||
let mut table = self.blocks.write().await;
|
||||
table.blocks = blocks;
|
||||
table.generation += 1;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Fetch one block's ciphertext, re-listing the revision once if its storage
|
||||
/// URL has expired.
|
||||
///
|
||||
/// A block fetch carries a `pm-storage-token` and no session credential at
|
||||
/// all, so it cannot fail for session reasons — only the URL expires. One
|
||||
/// re-list is the whole budget: a second expiry on a URL minted seconds ago
|
||||
/// is not an expiry, and replaying it would only loop.
|
||||
pub(crate) async fn ciphertext(&self, index: usize) -> Result<Bytes> {
|
||||
let (url, token, generation) = self.location(index).await?;
|
||||
match self.transport.http().get_storage_blob(&url, &token).await {
|
||||
Ok(bytes) => Ok(bytes),
|
||||
Err(e) if is_expired_block_url(&e) => {
|
||||
self.refresh(generation).await?;
|
||||
let (url, token, _) = self.location(index).await?;
|
||||
self.transport.http().get_storage_blob(&url, &token).await
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// An open handle on a file revision: its content key plus the block table and
|
||||
/// per-block plaintext sizes, resolved once by
|
||||
/// [`ProtonDriveClient::open_revision`](crate::ProtonDriveClient::open_revision).
|
||||
@@ -139,13 +258,10 @@ pub struct RevisionReader {
|
||||
/// can die under the reader — the renewal path. Deliberately *not* a whole
|
||||
/// client, so an anonymous public-link visitor can have one too.
|
||||
transport: RevisionTransport,
|
||||
uid: NodeUid,
|
||||
revision_id: String,
|
||||
content_key: ContentKey,
|
||||
blocks: RwLock<BlockTable>,
|
||||
/// Serializes block-table refreshes so a burst of expired-URL failures
|
||||
/// triggers one re-listing rather than one per block.
|
||||
refresh: Mutex<()>,
|
||||
/// The block table and its expired-URL recovery, shared with the whole-file
|
||||
/// download path.
|
||||
targets: BlockTargets,
|
||||
/// Plaintext size of each block, in block order.
|
||||
block_sizes: Vec<u64>,
|
||||
file_size: u64,
|
||||
@@ -161,16 +277,17 @@ impl RevisionReader {
|
||||
block_sizes: Vec<u64>,
|
||||
) -> Self {
|
||||
let file_size = block_sizes.iter().sum();
|
||||
Self {
|
||||
transport,
|
||||
let targets = BlockTargets::new(
|
||||
transport.clone(),
|
||||
uid,
|
||||
revision_id,
|
||||
content_key,
|
||||
blocks: RwLock::new(BlockTable {
|
||||
blocks,
|
||||
generation: 0,
|
||||
}),
|
||||
refresh: Mutex::new(()),
|
||||
block_sizes.len(),
|
||||
);
|
||||
Self {
|
||||
transport,
|
||||
content_key,
|
||||
targets,
|
||||
block_sizes,
|
||||
file_size,
|
||||
}
|
||||
@@ -178,12 +295,12 @@ impl RevisionReader {
|
||||
|
||||
/// The node this reader was opened on.
|
||||
pub fn uid(&self) -> &NodeUid {
|
||||
&self.uid
|
||||
self.targets.uid()
|
||||
}
|
||||
|
||||
/// The revision this reader is pinned to.
|
||||
pub fn revision_id(&self) -> &str {
|
||||
&self.revision_id
|
||||
self.targets.revision_id()
|
||||
}
|
||||
|
||||
/// Total plaintext size of the revision, summed from the block sizes
|
||||
@@ -259,71 +376,11 @@ impl RevisionReader {
|
||||
.await
|
||||
.map_err(|e| ProtonError::invalid_operation(format!("block slots closed: {e}")))?;
|
||||
|
||||
let (url, token, generation) = self.block_location(index).await?;
|
||||
|
||||
// A block fetch carries a `pm-storage-token` and no session credential
|
||||
// at all, so it cannot fail for session reasons — only the URL expires.
|
||||
let ciphertext = match self.transport.http().get_storage_blob(&url, &token).await {
|
||||
Ok(bytes) => bytes,
|
||||
Err(e) if is_expired_block_url(&e) => {
|
||||
self.refresh_blocks(generation).await?;
|
||||
let (url, token, _) = self.block_location(index).await?;
|
||||
self.transport.http().get_storage_blob(&url, &token).await?
|
||||
}
|
||||
Err(e) => return Err(e),
|
||||
};
|
||||
let ciphertext = self.targets.ciphertext(index).await?;
|
||||
|
||||
let plaintext = decrypt_block_blocking(self.content_key.clone(), ciphertext).await?;
|
||||
Ok((plaintext, permit))
|
||||
}
|
||||
|
||||
/// The storage URL and token for a block, plus the generation of the table
|
||||
/// they came from.
|
||||
async fn block_location(&self, index: usize) -> Result<(String, String, u64)> {
|
||||
let table = self.blocks.read().await;
|
||||
let block = table.blocks.get(index).ok_or_else(|| {
|
||||
ProtonError::invalid_operation(format!(
|
||||
"block {index} is missing from revision {}",
|
||||
self.revision_id
|
||||
))
|
||||
})?;
|
||||
Ok((
|
||||
block.bare_url.clone(),
|
||||
block.token.clone(),
|
||||
table.generation,
|
||||
))
|
||||
}
|
||||
|
||||
/// Re-list the revision to obtain fresh block URLs, unless another task has
|
||||
/// already replaced the generation the caller saw.
|
||||
async fn refresh_blocks(&self, seen_generation: u64) -> Result<()> {
|
||||
let _guard = self.refresh.lock().await;
|
||||
|
||||
if self.blocks.read().await.generation != seen_generation {
|
||||
// Someone else refreshed while we waited; their table is at least as
|
||||
// fresh as one we would fetch now.
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let (_, blocks) = self
|
||||
.transport
|
||||
.list_blocks(&self.uid.volume_id, &self.uid.link_id, &self.revision_id)
|
||||
.await?;
|
||||
|
||||
if blocks.len() != self.block_sizes.len() {
|
||||
return Err(ProtonError::invalid_operation(format!(
|
||||
"revision {} changed block count while open ({} -> {})",
|
||||
self.revision_id,
|
||||
self.block_sizes.len(),
|
||||
blocks.len()
|
||||
)));
|
||||
}
|
||||
|
||||
let mut table = self.blocks.write().await;
|
||||
table.blocks = blocks;
|
||||
table.generation += 1;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Which blocks overlap the plaintext range `[offset, end)`, paired with where
|
||||
@@ -468,4 +525,33 @@ mod tests {
|
||||
"the size stat would show"
|
||||
);
|
||||
}
|
||||
|
||||
/// What [`BlockTargets::ciphertext`] treats as "this URL expired, re-list
|
||||
/// the revision once". A block fetch carries no session credential, so a
|
||||
/// 401/403 there cannot mean our session — it means the URL's own
|
||||
/// authorization lapsed. Anything else is a real failure and is returned
|
||||
/// unchanged rather than costing a listing round trip.
|
||||
#[test]
|
||||
fn only_authorization_statuses_count_as_an_expired_block_url() {
|
||||
use proton_sdk::api::ResponseCode;
|
||||
use proton_sdk::error::ProtonApiError;
|
||||
|
||||
let api = |http_status: u16| {
|
||||
ProtonError::Api(ProtonApiError {
|
||||
code: ResponseCode::Unknown,
|
||||
http_status,
|
||||
message: String::new(),
|
||||
details: None,
|
||||
})
|
||||
};
|
||||
for status in [401, 403, 404] {
|
||||
assert!(is_expired_block_url(&api(status)), "http {status}");
|
||||
}
|
||||
for status in [400, 429, 500, 503] {
|
||||
assert!(!is_expired_block_url(&api(status)), "http {status}");
|
||||
}
|
||||
assert!(!is_expired_block_url(&ProtonError::invalid_operation(
|
||||
"bug"
|
||||
)));
|
||||
}
|
||||
}
|
||||
@@ -28,16 +28,25 @@ pub const DEFAULT_MAX_RETRIES: u32 = 3;
|
||||
/// Default base delay for exponential backoff between retries.
|
||||
pub const DEFAULT_RETRY_BASE_DELAY: Duration = Duration::from_millis(500);
|
||||
|
||||
/// Default upper bound for a single backoff sleep. A server-supplied
|
||||
/// `Retry-After` is honoured even if it exceeds this.
|
||||
/// Default upper bound for a single computed backoff sleep. A server-supplied
|
||||
/// `Retry-After` overrides it, up to [`DEFAULT_MAX_RETRY_AFTER`].
|
||||
pub const DEFAULT_RETRY_MAX_DELAY: Duration = Duration::from_secs(30);
|
||||
|
||||
/// Default upper bound for a server-supplied `Retry-After` wait.
|
||||
///
|
||||
/// The header is advice, and a caller is blocked on the sleep for its whole
|
||||
/// duration: an unbounded one is indistinguishable, from outside, from a hung
|
||||
/// process. Waiting less than asked risks another rate-limit answer, which the
|
||||
/// retry budget absorbs; waiting an hour risks a daemon that looks dead.
|
||||
pub const DEFAULT_MAX_RETRY_AFTER: Duration = Duration::from_secs(120);
|
||||
|
||||
/// Controls automatic retries on retryable responses (HTTP 408/429/502/503/504)
|
||||
/// and transient transport errors (timeout, connect).
|
||||
///
|
||||
/// Mirrors the Polly-style retry pipeline in the C# SDK: a server-supplied
|
||||
/// `Retry-After` header wins; otherwise the delay is exponential backoff
|
||||
/// (`base_delay * 2^attempt`, capped at `max_delay`) with full jitter.
|
||||
/// `Retry-After` header wins (clamped to `max_retry_after`); otherwise the delay
|
||||
/// is exponential backoff (`base_delay * 2^attempt`, capped at `max_delay`) with
|
||||
/// full jitter.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct RetryPolicy {
|
||||
/// Maximum number of retries after the initial attempt. `0` disables retry.
|
||||
@@ -46,6 +55,9 @@ pub struct RetryPolicy {
|
||||
pub base_delay: Duration,
|
||||
/// Cap on a single computed backoff sleep.
|
||||
pub max_delay: Duration,
|
||||
/// Cap on a server-supplied `Retry-After` wait. A longer header value is
|
||||
/// clamped to this.
|
||||
pub max_retry_after: Duration,
|
||||
}
|
||||
|
||||
impl Default for RetryPolicy {
|
||||
@@ -54,6 +66,7 @@ impl Default for RetryPolicy {
|
||||
max_retries: DEFAULT_MAX_RETRIES,
|
||||
base_delay: DEFAULT_RETRY_BASE_DELAY,
|
||||
max_delay: DEFAULT_RETRY_MAX_DELAY,
|
||||
max_retry_after: DEFAULT_MAX_RETRY_AFTER,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -32,6 +32,24 @@ impl ProtonError {
|
||||
pub fn invalid_operation(message: impl Into<String>) -> Self {
|
||||
Self::InvalidOperation(message.into())
|
||||
}
|
||||
|
||||
/// Whether replaying the request that produced this error could plausibly
|
||||
/// succeed (C# `RetryPolicy.IsRetriable`).
|
||||
///
|
||||
/// Used by the retry loops that sit *above* the HTTP client — block upload
|
||||
/// and block download — which retry a whole prepared transfer rather than a
|
||||
/// single request, and so cannot rely on [`crate::http`]'s own policy. A
|
||||
/// permanent rejection of the revision must fail the transfer at once
|
||||
/// instead of being replayed until the attempt budget runs out.
|
||||
pub fn is_retriable(&self) -> bool {
|
||||
match self {
|
||||
Self::Api(error) => error.is_retriable(),
|
||||
// A serialization or crypto failure is deterministic; an invariant
|
||||
// violation is a bug. Only a transport failure is worth replaying.
|
||||
Self::Transport(_) => true,
|
||||
Self::Serialization(_) | Self::Crypto(_) | Self::InvalidOperation(_) => false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// An error reported by the Proton API in its response envelope.
|
||||
@@ -50,6 +68,23 @@ pub struct ProtonApiError {
|
||||
}
|
||||
|
||||
impl ProtonApiError {
|
||||
/// Whether replaying the request could plausibly succeed.
|
||||
///
|
||||
/// A 4xx cannot succeed on a replay, except 408 and 429, and 404 — which is
|
||||
/// retriable only because both block-transfer callers re-request the
|
||||
/// transfer target first (C# `RetryPolicy.StatusCodeIsRetriable`). The
|
||||
/// envelope code is consulted as well as the HTTP status, because a rate
|
||||
/// limit can arrive as `Code: 429` under a 2xx-shaped envelope.
|
||||
pub fn is_retriable(&self) -> bool {
|
||||
if matches!(
|
||||
self.code,
|
||||
ResponseCode::TooManyRequests | ResponseCode::DoesNotExist
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
!(400..500).contains(&self.http_status) || matches!(self.http_status, 404 | 408 | 429)
|
||||
}
|
||||
|
||||
pub fn is_unauthorized(&self) -> bool {
|
||||
self.http_status == 401
|
||||
}
|
||||
@@ -81,3 +116,55 @@ impl ProtonApiError {
|
||||
serde_json::from_value(self.details.clone()?).ok()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn api(http_status: u16, code: ResponseCode) -> ProtonError {
|
||||
ProtonError::Api(ProtonApiError {
|
||||
code,
|
||||
http_status,
|
||||
message: String::new(),
|
||||
details: None,
|
||||
})
|
||||
}
|
||||
|
||||
/// The retriable set a block transfer may replay: transient statuses, plus
|
||||
/// 404, which means the transfer target expired and the caller re-mints it.
|
||||
#[test]
|
||||
fn transient_statuses_and_404_are_retriable() {
|
||||
for status in [404, 408, 429, 500, 502, 503, 504] {
|
||||
assert!(
|
||||
api(status, ResponseCode::Unknown).is_retriable(),
|
||||
"http {status} should be retriable"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Everything else in the 4xx family is a permanent rejection: replaying it
|
||||
/// only spends the attempt budget.
|
||||
#[test]
|
||||
fn other_client_errors_are_not_retriable() {
|
||||
for status in [400, 401, 403, 409, 422] {
|
||||
assert!(
|
||||
!api(status, ResponseCode::Unknown).is_retriable(),
|
||||
"http {status} should not be retriable"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// A rate limit can arrive as an envelope code under a 2xx-shaped response,
|
||||
/// and a missing block as `DoesNotExist`; both still mean "try again".
|
||||
#[test]
|
||||
fn envelope_codes_override_a_permanent_status() {
|
||||
assert!(api(400, ResponseCode::TooManyRequests).is_retriable());
|
||||
assert!(api(422, ResponseCode::DoesNotExist).is_retriable());
|
||||
}
|
||||
|
||||
/// Non-API failures: only a transport error is worth replaying.
|
||||
#[test]
|
||||
fn only_transport_failures_retry_off_the_api_path() {
|
||||
assert!(!ProtonError::invalid_operation("bug").is_retriable());
|
||||
}
|
||||
}
|
||||
@@ -738,8 +738,9 @@ fn api_error(status: StatusCode, bytes: &[u8]) -> ProtonError {
|
||||
/// `multipart` can't be cloned), so every retry resends the full request.
|
||||
///
|
||||
/// Retryable = HTTP 408/429/502/503/504 or a transient transport error
|
||||
/// (timeout / connect). A `Retry-After` header (delta-seconds) is honoured;
|
||||
/// otherwise the delay is exponential backoff with full jitter. Non-retryable
|
||||
/// (timeout / connect). A `Retry-After` header (delta-seconds) is honoured, up
|
||||
/// to `policy.max_retry_after`; otherwise the delay is exponential backoff with
|
||||
/// full jitter. Non-retryable
|
||||
/// responses and errors — including ordinary 4xx and the 401 that drives token
|
||||
/// refresh — pass straight through to the caller untouched.
|
||||
async fn send_retrying<F>(policy: &RetryPolicy, build: F) -> Result<reqwest::Response>
|
||||
@@ -751,7 +752,9 @@ where
|
||||
match build().send().await {
|
||||
Ok(response) => {
|
||||
if attempt < policy.max_retries && is_retryable_status(response.status()) {
|
||||
let delay = retry_after(&response).unwrap_or_else(|| backoff(policy, attempt));
|
||||
let delay = retry_after(&response)
|
||||
.map(|delay| delay.min(policy.max_retry_after))
|
||||
.unwrap_or_else(|| backoff(policy, attempt));
|
||||
tokio::time::sleep(delay).await;
|
||||
attempt += 1;
|
||||
continue;
|
||||
@@ -852,12 +855,28 @@ mod tests {
|
||||
assert_eq!(parse_retry_after_secs(""), None);
|
||||
}
|
||||
|
||||
/// A `Retry-After` longer than the policy allows is clamped: the caller is
|
||||
/// blocked on this sleep, and an unbounded one looks exactly like a hang.
|
||||
#[test]
|
||||
fn retry_after_is_clamped_to_policy_ceiling() {
|
||||
let policy = RetryPolicy::default();
|
||||
let advised = parse_retry_after_secs("3600").expect("delta-seconds parses");
|
||||
assert_eq!(
|
||||
advised.min(policy.max_retry_after),
|
||||
crate::config::DEFAULT_MAX_RETRY_AFTER
|
||||
);
|
||||
// A shorter wait is honoured verbatim.
|
||||
let short = parse_retry_after_secs("5").expect("delta-seconds parses");
|
||||
assert_eq!(short.min(policy.max_retry_after), Duration::from_secs(5));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn backoff_grows_then_caps_within_jitter_bounds() {
|
||||
let policy = RetryPolicy {
|
||||
max_retries: 5,
|
||||
base_delay: Duration::from_millis(100),
|
||||
max_delay: Duration::from_millis(1000),
|
||||
..RetryPolicy::default()
|
||||
};
|
||||
// Full jitter: every sample stays within [0, ceiling] where the
|
||||
// ceiling is base*2^attempt capped at max_delay.
|
||||
|
||||
Reference in new issue
Block a user