From ee16f4326e065db98b3149e94e8cb6354a3ef79c Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Sun, 25 Oct 2026 20:33:29 +0100 Subject: [PATCH] feat(http): send a read again when its connection drops A GET, or a POST that only reads, is now retried when the connection drops after the request left: before the response arrives or partway through its body. A phone does this when the OS closes a pooled connection or moves from Wi-Fi to mobile data, and until now the SDK only retried timeouts and failed connects, so every caller that walks a tree needed its own retry loop. send_retrying reads the body inside the loop and takes a Replay mode. Writes keep the old rule: a request that may have reached the server is not sent again, as a second attempt could create a duplicate. ApiHttpClient::post_query is the POST for endpoints that only read. Link details, thumbnail URLs, photo duplicates and checkAvailableHashes use it, on both the authenticated and the public-link paths. Block downloads from storage are retried as reads too. --- crates/proton-drive-rs/src/client.rs | 15 +- crates/proton-drive-rs/src/public_link.rs | 6 +- crates/proton-drive-rs/src/transport.rs | 10 +- crates/proton-sdk/src/http.rs | 366 ++++++++++++++++------ 4 files changed, 292 insertions(+), 105 deletions(-) diff --git a/crates/proton-drive-rs/src/client.rs b/crates/proton-drive-rs/src/client.rs index ef58fb1..1a3e667 100644 --- a/crates/proton-drive-rs/src/client.rs +++ b/crates/proton-drive-rs/src/client.rs @@ -1828,7 +1828,7 @@ impl ProtonDriveClient { let response: ThumbnailBlockListResponse = self .http - .post( + .post_query( &format!("volumes/{}/thumbnails", uid.volume_id), &ThumbnailBlockListRequest { thumbnail_ids: vec![thumbnail_id.clone()], @@ -1925,7 +1925,7 @@ impl ProtonDriveClient { for chunk in thumbnail_ids.chunks(MAX_THUMBNAIL_IDS_PER_REQUEST) { let response: ThumbnailBlockListResponse = match self .http - .post( + .post_query( &format!("volumes/{volume_id}/thumbnails"), &ThumbnailBlockListRequest { thumbnail_ids: chunk.to_vec(), @@ -2638,7 +2638,8 @@ impl ProtonDriveClient { "v2/volumes/{}/links/{}/checkAvailableHashes", parent_uid.volume_id, parent_uid.link_id ); - let response: NodeNameAvailabilityResponse = self.http.post(&path, &request).await?; + let response: NodeNameAvailabilityResponse = + self.http.post_query(&path, &request).await?; // Take the first hash the server reports free and map it back to its // name (C# returns `AvailableHashes[0]`). @@ -5855,7 +5856,7 @@ impl ProtonDriveClient { let path = format!("volumes/{}/photos/duplicates", photos_root.volume_id); let response: FindPhotoDuplicatesResponse = self .http - .post( + .post_query( &path, &FindPhotoDuplicatesRequest { name_hashes: vec![name_hash.clone()], @@ -5959,7 +5960,7 @@ impl ProtonDriveClient { let path = format!("volumes/{}/photos/duplicates", photos_root.volume_id); let response: FindPhotoDuplicatesResponse = self .http - .post( + .post_query( &path, &FindPhotoDuplicatesRequest { name_hashes: chunk.to_vec(), @@ -6045,7 +6046,7 @@ impl ProtonDriveClient { let path = format!("volumes/{}/photos/duplicates", photos_root.volume_id); let response: FindPhotoDuplicatesResponse = self .http - .post( + .post_query( &path, &FindPhotoDuplicatesRequest { name_hashes: chunk.to_vec(), @@ -7464,7 +7465,7 @@ impl ProtonDriveClient { format!("v2/volumes/{volume_id}/links") }; let body = LinkDetailsRequest { link_ids }; - self.http.post(&path, &body).await + self.http.post_query(&path, &body).await } async fn build_node( diff --git a/crates/proton-drive-rs/src/public_link.rs b/crates/proton-drive-rs/src/public_link.rs index 0302f98..4dad737 100644 --- a/crates/proton-drive-rs/src/public_link.rs +++ b/crates/proton-drive-rs/src/public_link.rs @@ -587,7 +587,7 @@ impl ProtonDrivePublicLinkClient { link_ids: &[LinkId], ) -> Result { self.transport - .post( + .post_query( &format!("v2/volumes/{volume_id}/links"), &LinkDetailsRequest { link_ids }, ) @@ -794,7 +794,7 @@ impl ProtonDrivePublicLinkClient { let response: ThumbnailBlockListResponse = self .transport - .post( + .post_query( &format!("volumes/{}/thumbnails", uid.volume_id), &ThumbnailBlockListRequest { thumbnail_ids: vec![thumbnail_id.clone()], @@ -880,7 +880,7 @@ impl ProtonDrivePublicLinkClient { ) { let response: ThumbnailBlockListResponse = match self .transport - .post( + .post_query( &format!("volumes/{volume_id}/thumbnails"), &ThumbnailBlockListRequest { thumbnail_ids: chunk.to_vec(), diff --git a/crates/proton-drive-rs/src/transport.rs b/crates/proton-drive-rs/src/transport.rs index 91dc6a8..78914e4 100644 --- a/crates/proton-drive-rs/src/transport.rs +++ b/crates/proton-drive-rs/src/transport.rs @@ -127,17 +127,19 @@ impl RevisionTransport { } } - /// `POST path` through the current session, with the same one-shot renewal. - pub(crate) async fn post( + /// `POST path` for an endpoint that only reads (see + /// [`ApiHttpClient::post_query`]), through the current session, with the + /// same one-shot renewal. + pub(crate) async fn post_query( &self, path: &str, body: &B, ) -> Result { let (http, generation) = self.session.http(); - match http.post::(path, body).await { + match http.post_query::(path, body).await { Err(e) if is_session_expired(&e) => { self.session.renew(generation).await?; - self.session.http().0.post::(path, body).await + self.session.http().0.post_query::(path, body).await } other => other, } diff --git a/crates/proton-sdk/src/http.rs b/crates/proton-sdk/src/http.rs index ef1410d..14ca879 100644 --- a/crates/proton-sdk/src/http.rs +++ b/crates/proton-sdk/src/http.rs @@ -239,25 +239,24 @@ impl ApiHttpClient { /// blocking decrypt task. pub async fn get_storage_blob(&self, url: &str, token: &str) -> Result { let mut timer = self.telemetry().start("storage_download"); - let response = send_retrying(&self.inner.config.retry_policy, || { - // Override the client-level (API) timeout: a 4 MiB block on a slow - // uplink legitimately outruns the 30s JSON budget. - let mut request = self - .inner - .http - .get(url) - .timeout(self.inner.config.storage_timeout) - .header(STORAGE_TOKEN_HEADER, token); - if !self.inner.config.user_agent.is_empty() { - request = - request.header(reqwest::header::USER_AGENT, &self.inner.config.user_agent); - } - request - }) - .await?; - let status = response.status(); + let Buffered { status, bytes, .. } = + send_retrying(&self.inner.config.retry_policy, Replay::Safe, || { + // Override the client-level (API) timeout: a 4 MiB block on a slow + // uplink legitimately outruns the 30s JSON budget. + let mut request = self + .inner + .http + .get(url) + .timeout(self.inner.config.storage_timeout) + .header(STORAGE_TOKEN_HEADER, token); + if !self.inner.config.user_agent.is_empty() { + request = + request.header(reqwest::header::USER_AGENT, &self.inner.config.user_agent); + } + request + }) + .await?; timer.attr("status", status.as_u16()); - let bytes = response.bytes().await?; // Success bodies are raw block bytes (not JSON); only error responses // carry the envelope. @@ -294,32 +293,31 @@ impl ApiHttpClient { 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 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"); - let form = reqwest::multipart::Form::new().part("Block", part); + let Buffered { status, bytes, .. } = + send_retrying(&self.inner.config.retry_policy, Replay::Unsafe, || { + 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"); + let form = reqwest::multipart::Form::new().part("Block", part); - let mut request = self - .inner - .http - .post(url) - .timeout(self.inner.config.storage_timeout) - .header(STORAGE_TOKEN_HEADER, token) - .multipart(form); + let mut request = self + .inner + .http + .post(url) + .timeout(self.inner.config.storage_timeout) + .header(STORAGE_TOKEN_HEADER, token) + .multipart(form); - if !self.inner.config.user_agent.is_empty() { - request = - request.header(reqwest::header::USER_AGENT, &self.inner.config.user_agent); - } - request - }) - .await?; - let status = response.status(); + if !self.inner.config.user_agent.is_empty() { + request = + request.header(reqwest::header::USER_AGENT, &self.inner.config.user_agent); + } + request + }) + .await?; timer.attr("status", status.as_u16()); - let bytes = response.bytes().await?; if let Ok(envelope) = serde_json::from_slice::(&bytes) { if !envelope.is_success() { @@ -338,12 +336,29 @@ impl ApiHttpClient { /// `GET {path}` returning a typed success body. pub async fn get(&self, path: &str) -> Result { - self.send::<(), T>(Method::GET, path, None).await + self.send::<(), T>(Method::GET, path, None, Replay::Safe) + .await } /// `POST {path}` with a JSON body, returning a typed success body. pub async fn post(&self, path: &str, body: &B) -> Result { - self.send::(Method::POST, path, Some(body)).await + self.send::(Method::POST, path, Some(body), Replay::Unsafe) + .await + } + + /// `POST {path}` for an endpoint that only reads, such as fetching link + /// details or thumbnail URLs, where the body is a query too long for a URL. + /// + /// Retried like a `GET`: a connection that drops after the request left is + /// sent again, because the server changed nothing. Only use this for an + /// endpoint that changes nothing; [`post`](Self::post) for everything else. + pub async fn post_query( + &self, + path: &str, + body: &B, + ) -> Result { + self.send::(Method::POST, path, Some(body), Replay::Safe) + .await } /// `POST {path}` as multipart form data with a JSON `Metadata` part and @@ -361,15 +376,15 @@ impl ApiHttpClient { let response = self .send_multipart_with_token(path, &metadata, binary_parts, &rejected_token) .await?; - let response = if response.status() == StatusCode::UNAUTHORIZED && self.refresh_tokens { - let bytes = response.bytes().await?; - if let Ok(envelope) = serde_json::from_slice::(&bytes) + let response = if response.status == StatusCode::UNAUTHORIZED && self.refresh_tokens { + let bytes = &response.bytes; + if let Ok(envelope) = serde_json::from_slice::(bytes) && matches!( envelope.code, ResponseCode::AccountDeleted | ResponseCode::AccountDisabled ) { - return Err(api_error(StatusCode::UNAUTHORIZED, &bytes)); + return Err(api_error(StatusCode::UNAUTHORIZED, bytes)); } let token = self.refresh_access_token(&rejected_token).await?; self.send_multipart_with_token(path, &metadata, binary_parts, &token) @@ -377,20 +392,22 @@ impl ApiHttpClient { } else { response }; - timer.attr("status", response.status().as_u16()); - let parsed = parse_response(response).await?; + timer.attr("status", response.status.as_u16()); + let parsed = parse_buffered(response)?; timer.success(); Ok(parsed) } /// `PUT {path}` with a JSON body, returning a typed success body. pub async fn put(&self, path: &str, body: &B) -> Result { - self.send::(Method::PUT, path, Some(body)).await + self.send::(Method::PUT, path, Some(body), Replay::Unsafe) + .await } /// `DELETE {path}` returning a typed success body. pub async fn delete(&self, path: &str) -> Result { - self.send::<(), T>(Method::DELETE, path, None).await + self.send::<(), T>(Method::DELETE, path, None, Replay::Unsafe) + .await } /// `DELETE {path}` with a JSON body, returning a typed success body. @@ -403,7 +420,8 @@ impl ApiHttpClient { path: &str, body: &B, ) -> Result { - self.send::(Method::DELETE, path, Some(body)).await + self.send::(Method::DELETE, path, Some(body), Replay::Unsafe) + .await } async fn send( @@ -411,6 +429,7 @@ impl ApiHttpClient { method: Method, path: &str, body: Option<&B>, + replay: Replay, ) -> Result { let mut timer = self.telemetry().start("http_request"); timer.attr("method", method.as_str()); @@ -419,18 +438,18 @@ impl ApiHttpClient { // An early `?` here records the op as a failure (OpTimer defaults to it). let response = self - .send_with_token(method.clone(), path, body, &access_token) + .send_with_token(method.clone(), path, body, replay, &access_token) .await?; - let response = if response.status() == StatusCode::UNAUTHORIZED && self.refresh_tokens { - self.handle_unauthorized(method, path, body, response, access_token) + let response = if response.status == StatusCode::UNAUTHORIZED && self.refresh_tokens { + self.handle_unauthorized(method, path, body, replay, response, access_token) .await? } else { response }; - timer.attr("status", response.status().as_u16()); - let parsed = parse_response(response).await?; + timer.attr("status", response.status.as_u16()); + let parsed = parse_buffered(response)?; timer.success(); Ok(parsed) } @@ -440,22 +459,22 @@ impl ApiHttpClient { method: Method, path: &str, body: Option<&B>, - response: reqwest::Response, + replay: Replay, + response: Buffered, rejected_access_token: String, - ) -> Result { + ) -> Result { // Don't bother refreshing for terminal account states. - let bytes = response.bytes().await?; - if let Ok(envelope) = serde_json::from_slice::(&bytes) + if let Ok(envelope) = serde_json::from_slice::(&response.bytes) && matches!( envelope.code, ResponseCode::AccountDeleted | ResponseCode::AccountDisabled ) { - return Err(api_error(StatusCode::UNAUTHORIZED, &bytes)); + return Err(api_error(StatusCode::UNAUTHORIZED, &response.bytes)); } let access_token = self.refresh_access_token(&rejected_access_token).await?; - self.send_with_token(method, path, body, &access_token) + self.send_with_token(method, path, body, replay, &access_token) .await } @@ -464,15 +483,16 @@ impl ApiHttpClient { method: Method, path: &str, body: Option<&B>, + replay: Replay, access_token: &str, - ) -> Result { + ) -> Result { let url = format!( "{}{}{}", self.inner.base_url, self.route_prefix, path.trim_start_matches('/') ); - send_retrying(&self.inner.config.retry_policy, || { + send_retrying(&self.inner.config.retry_policy, replay, || { let mut request = self .inner .http @@ -502,14 +522,14 @@ impl ApiHttpClient { metadata: &[u8], binary_parts: &[(String, Vec)], access_token: &str, - ) -> Result { + ) -> Result { let url = format!( "{}{}{}", self.inner.base_url, self.route_prefix, path.trim_start_matches('/') ); - send_retrying(&self.inner.config.retry_policy, || { + send_retrying(&self.inner.config.retry_policy, Replay::Unsafe, || { let metadata_part = reqwest::multipart::Part::bytes(metadata.to_vec()) .file_name("Metadata") .mime_str("application/json") @@ -652,7 +672,7 @@ pub async fn get_unauthenticated( let base_url = ensure_trailing_slash(&config.base_url); let url = format!("{}{}", base_url, path.trim_start_matches('/')); - let response = send_retrying(&config.retry_policy, || { + let response = send_retrying(&config.retry_policy, Replay::Safe, || { let mut request = http .get(&url) .header(APP_VERSION_HEADER, &config.app_version) @@ -664,7 +684,7 @@ pub async fn get_unauthenticated( }) .await?; - parse_response(response).await + parse_buffered(response) } /// `POST {path}` without a session: no `x-pm-uid` and no bearer token. @@ -699,7 +719,7 @@ pub async fn post_unauthenticated_verified( let base_url = ensure_trailing_slash(&config.base_url); let url = format!("{}{}", base_url, path.trim_start_matches('/')); - let response = send_retrying(&config.retry_policy, || { + let response = send_retrying(&config.retry_policy, Replay::Unsafe, || { let mut request = http .post(&url) .header(APP_VERSION_HEADER, &config.app_version) @@ -717,15 +737,22 @@ pub async fn post_unauthenticated_verified( }) .await?; - parse_response(response).await + parse_buffered(response) } /// Read a response body, enforce the Proton success envelope, and deserialize /// the typed success payload. async fn parse_response(response: reqwest::Response) -> Result { - let status = response.status(); - let path = response.url().path().to_owned(); - let bytes = response.bytes().await?; + parse_buffered(Buffered::read(response).await?) +} + +/// [`parse_response`] for a response whose body is already read. +fn parse_buffered(response: Buffered) -> Result { + let Buffered { + status, + path, + bytes, + } = response; // Every Proton response embeds the envelope; a missing/non-success code or a // non-2xx HTTP status is an API error. `MultipleResponses` (1001) is a batch @@ -784,25 +811,67 @@ fn api_error(status: StatusCode, bytes: &[u8]) -> ProtonError { }) } -/// Send a request, transparently retrying retryable failures per `policy`. +/// Whether a request may be sent again after it may already have reached the +/// server. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Replay { + /// The request only reads: a `GET`, or a [`ApiHttpClient::post_query`]. + /// Sending it twice changes nothing, so a connection that drops after the + /// request left — mid-request or mid-body — is worth another attempt. + Safe, + /// The request may change something. Only a failure that shows the request + /// never reached the server is retried, plus a timeout (see + /// [`is_retryable_error`]). + Unsafe, +} + +/// A response with its body read to the end. +/// +/// The body is read inside [`send_retrying`], so a body that breaks off can be +/// retried for a [`Replay::Safe`] request like a failed send. +struct Buffered { + status: StatusCode, + /// The request path, for the failure log in [`parse_buffered`]. + path: String, + bytes: Bytes, +} + +impl Buffered { + async fn read(response: reqwest::Response) -> reqwest::Result { + let status = response.status(); + let path = response.url().path().to_owned(); + let bytes = response.bytes().await?; + Ok(Self { + status, + path, + bytes, + }) + } +} + +/// Send a request and read its body, transparently retrying retryable failures +/// per `policy`. /// /// `build` is called once per attempt to produce a fresh `RequestBuilder` /// (reqwest builders are consumed by `send`, and streaming bodies like /// `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, up +/// (timeout / connect). A [`Replay::Safe`] request is also retried when the +/// connection drops after the request was sent, before or during the body — +/// a phone does that when the OS closes a pooled connection or the network +/// changes. 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 +async fn send_retrying(policy: &RetryPolicy, replay: Replay, build: F) -> Result where F: Fn() -> reqwest::RequestBuilder, { let mut attempt: u32 = 0; loop { - match build().send().await { + let retry = match build().send().await { Ok(response) => { if attempt < policy.max_retries && is_retryable_status(response.status()) { let delay = retry_after(&response) @@ -812,16 +881,20 @@ where attempt += 1; continue; } - return Ok(response); - } - Err(err) => { - if attempt < policy.max_retries && is_retryable_error(&err) { - tokio::time::sleep(backoff(policy, attempt)).await; - attempt += 1; - continue; + match Buffered::read(response).await { + Ok(buffered) => return Ok(buffered), + // The server answered, so an unsafe request may have acted. + Err(err) => (replay == Replay::Safe, err), } - return Err(err.into()); } + Err(err) => (is_retryable_error(&err, replay), err), + }; + match retry { + (true, _) if attempt < policy.max_retries => { + tokio::time::sleep(backoff(policy, attempt)).await; + attempt += 1; + } + (_, err) => return Err(err.into()), } } } @@ -856,11 +929,12 @@ fn is_retryable_status(status: StatusCode) -> bool { matches!(status.as_u16(), 408 | 429 | 502 | 503 | 504) } -/// A transport error worth retrying: a timeout or a failure to connect. A -/// mid-body error (`is_body`) is not retried — the request may have been -/// applied server-side. -fn is_retryable_error(err: &reqwest::Error) -> bool { - err.is_timeout() || err.is_connect() +/// A failed send worth retrying: a timeout or a failure to connect, and for a +/// [`Replay::Safe`] request any failure of the request itself, such as a +/// connection closed before the response arrived. That last one is not retried +/// for an unsafe request — the request may have been applied server-side. +fn is_retryable_error(err: &reqwest::Error, replay: Replay) -> bool { + err.is_timeout() || err.is_connect() || (replay == Replay::Safe && err.is_request()) } /// Parse a `Retry-After` header expressed as delta-seconds. The HTTP-date form @@ -1358,6 +1432,116 @@ mod tests { server.abort(); } + /// How the first connection to [`dropping_server`] fails. + #[derive(Clone, Copy)] + enum Drop { + /// Read the request, then close without answering. + BeforeResponse, + /// Send the headers and part of the body, then close. + MidBody, + } + + /// A test server whose first connection fails as `drop` says and whose + /// later connections answer with success. Returns its address and how many + /// connections it accepted. + async fn dropping_server( + drop: Drop, + ) -> ( + std::net::SocketAddr, + Arc, + tokio::task::JoinHandle<()>, + ) { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let connections = Arc::new(AtomicUsize::new(0)); + let counted = connections.clone(); + let server = tokio::spawn(async move { + loop { + let (mut sock, _) = listener.accept().await.unwrap(); + let first = counted.fetch_add(1, Ordering::SeqCst) == 0; + let mut buf = [0u8; 2048]; + let _ = sock.read(&mut buf).await; + let body = br#"{"Code":1000}"#; + if first { + if let Drop::MidBody = drop { + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n", + body.len() + 100 + ); + let _ = sock.write_all(head.as_bytes()).await; + let _ = sock.write_all(&body[..4]).await; + let _ = sock.flush().await; + } + continue; + } + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + let _ = sock.write_all(head.as_bytes()).await; + let _ = sock.write_all(body).await; + let _ = sock.flush().await; + } + }); + (addr, connections, server) + } + + fn quick_retries() -> RetryPolicy { + RetryPolicy { + max_retries: 3, + base_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), + max_retry_after: Duration::from_millis(1), + } + } + + /// A read is sent again when the connection drops after the request left, + /// whether before the response or partway through its body. + #[tokio::test] + async fn a_read_is_sent_again_after_the_connection_drops() { + use std::sync::atomic::Ordering; + + for drop in [Drop::BeforeResponse, Drop::MidBody] { + let (addr, connections, server) = dropping_server(drop).await; + let client = refresh_client(addr, quick_retries()); + let _: ApiResponse = client.get("listing").await.expect("the GET is retried"); + assert_eq!(connections.load(Ordering::SeqCst), 2); + server.abort(); + + let (addr, connections, server) = dropping_server(drop).await; + let client = refresh_client(addr, quick_retries()); + let _: ApiResponse = client + .post_query("links", &serde_json::json!({})) + .await + .expect("the read-only POST is retried"); + assert_eq!(connections.load(Ordering::SeqCst), 2); + server.abort(); + } + } + + /// A write is not sent again once the request may have reached the server: + /// the first attempt may have acted, and a second one could create a + /// duplicate. + #[tokio::test] + async fn a_write_is_not_sent_again_after_the_connection_drops() { + use std::sync::atomic::Ordering; + + for drop in [Drop::BeforeResponse, Drop::MidBody] { + let (addr, connections, server) = dropping_server(drop).await; + let client = refresh_client(addr, quick_retries()); + let error = client + .post::<_, ApiResponse>("folders", &serde_json::json!({})) + .await + .expect_err("the POST is not retried"); + assert!(matches!(error, ProtonError::Transport(_)), "{error:?}"); + assert_eq!(connections.load(Ordering::SeqCst), 1); + server.abort(); + } + } + /// A 5xx from `auth/v4/refresh` may come after Proton rotated the tokens; /// a retry would spend the old refresh token a second time. #[tokio::test]