diff --git a/UPSTREAM_SYNC.md b/UPSTREAM_SYNC.md index a75e781..6a2305b 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**: `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`; 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` / `Option`, `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` — 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, 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.| diff --git a/crates/proton-drive-rs/src/client.rs b/crates/proton-drive-rs/src/client.rs index 2c48dca..e724fee 100644 --- a/crates/proton-drive-rs/src/client.rs +++ b/crates/proton-drive-rs/src/client.rs @@ -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( &self, - blocks: &[BlockDto], + targets: Arc, + block_count: usize, content_key: &ContentKey, manifest: &mut Vec, 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") + )); + } } diff --git a/crates/proton-drive-rs/src/public_link.rs b/crates/proton-drive-rs/src/public_link.rs index 246086c..0302f98 100644 --- a/crates/proton-drive-rs/src/public_link.rs +++ b/crates/proton-drive-rs/src/public_link.rs @@ -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)) } diff --git a/crates/proton-drive-rs/src/revision.rs b/crates/proton-drive-rs/src/revision.rs index 6f8a54f..74645b9 100644 --- a/crates/proton-drive-rs/src/revision.rs +++ b/crates/proton-drive-rs/src/revision.rs @@ -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, + /// 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, + 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 { + 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, - /// 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, file_size: u64, @@ -161,16 +277,17 @@ impl RevisionReader { block_sizes: Vec, ) -> Self { let file_size = block_sizes.iter().sum(); - Self { - transport, + let targets = BlockTargets::new( + transport.clone(), uid, revision_id, + blocks, + block_sizes.len(), + ); + Self { + transport, content_key, - blocks: RwLock::new(BlockTable { - blocks, - generation: 0, - }), - refresh: Mutex::new(()), + 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" + ))); + } } diff --git a/crates/proton-sdk/src/config.rs b/crates/proton-sdk/src/config.rs index e38c2e6..b0dfd19 100644 --- a/crates/proton-sdk/src/config.rs +++ b/crates/proton-sdk/src/config.rs @@ -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, } } } diff --git a/crates/proton-sdk/src/error.rs b/crates/proton-sdk/src/error.rs index 1878dd1..5d1fca5 100644 --- a/crates/proton-sdk/src/error.rs +++ b/crates/proton-sdk/src/error.rs @@ -32,6 +32,24 @@ impl ProtonError { pub fn invalid_operation(message: impl Into) -> 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()); + } +} diff --git a/crates/proton-sdk/src/http.rs b/crates/proton-sdk/src/http.rs index 6d206ca..96cc77e 100644 --- a/crates/proton-sdk/src/http.rs +++ b/crates/proton-sdk/src/http.rs @@ -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(policy: &RetryPolicy, build: F) -> Result @@ -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.