diff --git a/src/api/files.rs b/src/api/files.rs index d6692ec..6a6a135 100644 --- a/src/api/files.rs +++ b/src/api/files.rs @@ -79,13 +79,41 @@ impl MDRSConnection { .to_string_lossy() .nfc() .collect(); - let file_bytes = tokio::fs::read(file_path).await?; - let part = multipart::Part::bytes(file_bytes).file_name(file_name.clone()); - let form = multipart::Form::new() - .text("folder_id", folder_id.to_string()) - .part("file", part); + let build_form = || async { + let file_bytes = tokio::fs::read(file_path).await?; + let part = multipart::Part::bytes(file_bytes).file_name(file_name.clone()); + Ok::<_, anyhow::Error>( + multipart::Form::new() + .text("folder_id", folder_id.to_string()) + .part("file", part), + ) + }; let _permit = limiter.acquire().await?; - let resp = self.post_multipart("v3/files/", form).await?; + let sent_with = self.token.clone(); + let resp = self + .post_multipart("v3/files/", build_form().await?) + .await?; + if resp.status() != reqwest::StatusCode::UNAUTHORIZED { + if !resp.status().is_success() { + bail!("Upload failed: {}", resp.status()); + } + return Ok(()); + } + // The body reached the server, but it may have waited to be served for longer + // than the access token it was sent with lived. Re-reading the file and sending + // it again is only worth it if the token has actually moved on: any other reason + // for the refusal would just cost a second full transfer to be told the same. + let retry_conn = self.connection_with_fresh_token_for_retry().await?; + if retry_conn.token == sent_with { + bail!("Upload failed: {}", resp.status()); + } + let resp = retry_conn + .post_multipart("v3/files/", build_form().await?) + .await?; + if resp.status() == reqwest::StatusCode::UNAUTHORIZED { + // Refused twice, the second time with a token that was current when it left. + bail!(crate::connection::SERVER_BUSY_MESSAGE); + } if !resp.status().is_success() { bail!("Upload failed: {}", resp.status()); } diff --git a/src/api/users.rs b/src/api/users.rs index 6b455f4..321d242 100644 --- a/src/api/users.rs +++ b/src/api/users.rs @@ -6,7 +6,8 @@ use std::time::Duration; /// Bound the refresh on its own: the caller holds a lock that spans processes while this /// runs, so a provider that accepts the connection and then goes quiet would stall every -/// other request on this machine rather than just this one. +/// other request on this machine rather than just this one. Uploads are served by a +/// separate instance, so this one is not queued behind them and has no reason to be slow. const TOKEN_REFRESH_TIMEOUT: Duration = Duration::from_secs(30); /// Full API response shape from GET v3/users/current/ diff --git a/src/connection.rs b/src/connection.rs index 3f1726a..c1fb5ab 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -1,9 +1,14 @@ +use anyhow::bail; use reqwest::header::{ACCEPT, AUTHORIZATION, HeaderMap, HeaderValue, USER_AGENT}; use reqwest::{Client, Response}; use serde::Serialize; use std::sync::Arc; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; +/// Shown when a request was refused twice for a token that was current when it left. +pub const SERVER_BUSY_MESSAGE: &str = "The server took too long to start handling the request and may be overloaded. \ +Try again, or reduce the number of parallel transfers."; + fn build_user_agent() -> String { let info = os_info::get(); let mut parts = vec![info.os_type().to_string()]; @@ -90,6 +95,12 @@ impl MDRSConnection { } } + /// Same as `connection_with_fresh_token`, for callers that manage their own retry + /// because the request body cannot simply be sent twice. + pub async fn connection_with_fresh_token_for_retry(&self) -> Result { + self.connection_with_fresh_token().await + } + async fn connection_with_fresh_token(&self) -> Result { match (&self.remote, &self.token) { (Some(remote), Some(_)) => { @@ -122,56 +133,87 @@ impl MDRSConnection { headers } - pub async fn get(&self, path: &str) -> Result { + /// Send a request, and send it once more if the server found the access token + /// expired. A request can wait in the server's queue for longer than the token it + /// was sent with lives, so a refusal here need not mean the session is over. + /// + /// The retry goes back through `connection_with_fresh_token`, which re-reads the + /// cache under the lock: if another process refreshed while this request waited, + /// its token is reused rather than a second one being minted. + async fn send_with_retry(&self, build: F) -> Result + where + F: Fn(&MDRSConnection) -> reqwest::RequestBuilder, + { let conn = self.connection_with_fresh_token().await?; - Ok(conn - .client - .get(conn.build_url(path)) - .headers(conn.prepare_headers()) - .send() - .await?) + let response = build(&conn).send().await?; + if response.status() != reqwest::StatusCode::UNAUTHORIZED { + return Ok(response); + } + // Only a stale access token is worth a second attempt, and the sign of one is a + // different token coming back. A wrong folder password, an anonymous request, or + // a session that has really ended all leave it untouched, and those refusals + // belong to the caller to report. + let retry_conn = self.connection_with_fresh_token().await?; + if retry_conn.token == conn.token { + return Ok(response); + } + let response = build(&retry_conn).send().await?; + if response.status() == reqwest::StatusCode::UNAUTHORIZED { + // Refused twice, the second time with a token that was current when it left. + // The session is fine; the server is not starting requests before their + // credentials lapse, which is worth saying plainly rather than sending the + // user off to log in again. + bail!(SERVER_BUSY_MESSAGE); + } + Ok(response) + } + + pub async fn get(&self, path: &str) -> Result { + self.send_with_retry(|conn| { + conn.client + .get(conn.build_url(path)) + .headers(conn.prepare_headers()) + }) + .await } pub async fn get_with_query(&self, path: &str, query: &Q) -> Result where Q: Serialize + ?Sized, { - let conn = self.connection_with_fresh_token().await?; - Ok(conn - .client - .get(conn.build_url(path)) - .headers(conn.prepare_headers()) - .query(query) - .send() - .await?) + self.send_with_retry(|conn| { + conn.client + .get(conn.build_url(path)) + .headers(conn.prepare_headers()) + .query(query) + }) + .await } pub async fn get_url(&self, url: &str) -> Result { - let conn = self.connection_with_fresh_token().await?; - Ok(conn - .client - .get(if url.starts_with("http") { - url.to_string() - } else { - conn.build_url(url) - }) - .headers(conn.prepare_headers()) - .send() - .await?) + self.send_with_retry(|conn| { + conn.client + .get(if url.starts_with("http") { + url.to_string() + } else { + conn.build_url(url) + }) + .headers(conn.prepare_headers()) + }) + .await } pub async fn post_json(&self, path: &str, body: &B) -> Result where B: Serialize + ?Sized, { - let conn = self.connection_with_fresh_token().await?; - Ok(conn - .client - .post(conn.build_url(path)) - .headers(conn.prepare_headers()) - .json(body) - .send() - .await?) + self.send_with_retry(|conn| { + conn.client + .post(conn.build_url(path)) + .headers(conn.prepare_headers()) + .json(body) + }) + .await } pub async fn post_multipart( @@ -190,13 +232,12 @@ impl MDRSConnection { } pub async fn delete(&self, path: &str) -> Result { - let conn = self.connection_with_fresh_token().await?; - Ok(conn - .client - .delete(conn.build_url(path)) - .headers(conn.prepare_headers()) - .send() - .await?) + self.send_with_retry(|conn| { + conn.client + .delete(conn.build_url(path)) + .headers(conn.prepare_headers()) + }) + .await } pub async fn delete_with_query( @@ -207,13 +248,48 @@ impl MDRSConnection { where Q: Serialize + ?Sized, { - let conn = self.connection_with_fresh_token().await?; - Ok(conn - .client - .delete(conn.build_url(path)) - .headers(conn.prepare_headers()) - .query(query) - .send() - .await?) + self.send_with_retry(|conn| { + conn.client + .delete(conn.build_url(path)) + .headers(conn.prepare_headers()) + .query(query) + }) + .await + } +} + +#[cfg(test)] +mod retry_tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + /// A refusal that no new token could answer belongs to the caller: an anonymous + /// request, a wrong folder password and an ended session all look like this, and + /// sending the same thing twice would only repeat the refusal. + #[tokio::test] + async fn a_refusal_is_handed_back_when_the_token_cannot_change() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut buf = [0u8; 4096]; + let _ = stream.read(&mut buf).await.unwrap(); + stream + .write_all( + b"HTTP/1.1 401 Unauthorized\r\ncontent-length: 0\r\nconnection: close\r\n\r\n", + ) + .await + .unwrap(); + }); + + // No remote and no token, so nothing can be refreshed and the request must not + // be sent a second time. The stub answers once and would hang on a retry. + let conn = MDRSConnection::new(&format!("http://{addr}")); + let response = conn.get("v3/anything/").await.unwrap(); + server.await.unwrap(); + + assert_eq!(response.status(), reqwest::StatusCode::UNAUTHORIZED); } }