feat(auth): send a request again when its token lapsed in the queue

A request can wait to be served for longer than the access token it was
sent with lives, and comes back refused for a token that was valid when
it left. Uploads that take minutes make that wait ordinary.

- send once more when a refusal is answered by a different token, which
  is the only refusal a second attempt can help; a wrong folder
  password, an anonymous request and an ended session all leave the
  token untouched and stay with the caller who knows what they mean
- rebuild the upload and read the file again for its retry, but only
  once the token has moved on: any other refusal would cost a second
  full transfer to be told the same
- say the server may be overloaded when the second attempt is refused
  too, rather than sending the user off to log in over a session that
  is fine

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());
         }

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/

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, anyhow::Error> {
+        self.connection_with_fresh_token().await
+    }
+
     async fn connection_with_fresh_token(&self) -> Result<Self, anyhow::Error> {
         match (&self.remote, &self.token) {
             (Some(remote), Some(_)) => {
@@ -122,56 +133,87 @@ impl MDRSConnection {
         headers
     }
 
-    pub async fn get(&self, path: &str) -> Result<Response, anyhow::Error> {
+    /// 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<F>(&self, build: F) -> Result<Response, anyhow::Error>
+    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<Response, anyhow::Error> {
+        self.send_with_retry(|conn| {
+            conn.client
+                .get(conn.build_url(path))
+                .headers(conn.prepare_headers())
+        })
+        .await
     }
 
     pub async fn get_with_query<Q>(&self, path: &str, query: &Q) -> Result<Response, anyhow::Error>
     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<Response, anyhow::Error> {
-        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<B>(&self, path: &str, body: &B) -> Result<Response, anyhow::Error>
     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<Response, anyhow::Error> {
-        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<Q>(
@@ -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);
     }
 }