add support for upload speed / remaining in the cache upload step (#8081

)

### Description

When turbo is waiting for the cache, don't really expose any data to the
user. This PR modifies the cache put operation to take a stream and
introduces a new stream adapter that captures a running history of
writes using a ring buffer. These requests are stored in a hashmap which
is read by the spinner while waiting for shutdown, and a live number is
posted with the remaining bytes and current upload speed.

Also gracefully shuts down the upload phase when pressing ctrl-c

Before:
```
>>  ...Finishing writing to cache...                                                                                                                                                                                                                     WARNING  failed to contact remote cache: skipping HTTP Request, too many failures have occurred.
Last error: error sending request for url (https://vercel.com/api/v8/artifacts/1e1db6d9c17b138d): operation timed out
```
After:
```
>   ...Finishing writing to cache... (1.50GB remaining, 8.86MB/s)                                                                                                                                                                                        WARNING  the cache artifact for d6b10341668ff294 was too large to upload within the timeout  
```

https://asciinema.org/a/s9AsFiVHQRISc22Ve0YY6IlTQ

### Testing Instructions

TBD


Closes TURBO-2977

crates/turborepo-api-client/Cargo.toml

@@ -21,15 +21,18 @@ workspace = true
 
 [dependencies]
 anyhow = { workspace = true }
+bytes.workspace = true
 chrono = { workspace = true, features = ["serde"] }
 lazy_static = { workspace = true }
 regex = { workspace = true }
-reqwest = { workspace = true, features = ["json"] }
+reqwest = { workspace = true, features = ["json", "stream"] }
 rustc_version_runtime = "0.2.1"
 serde = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
+tokio-stream = "0.1.15"
+tokio-util = { version = "0.7.10", features = ["codec"] }
 tracing = { workspace = true }
 turbopath = { workspace = true }
 turborepo-ci = { workspace = true }

crates/turborepo-api-client/src/error.rs

@@ -7,6 +7,8 @@ use crate::CachingStatus;
 
 #[derive(Debug, Error)]
 pub enum Error {
+    #[error("Error reading from disk: {0}")]
+    ReadError(#[from] std::io::Error),
     #[error("Error making HTTP request: {0}")]
     ReqwestError(#[from] reqwest::Error),
     #[error("skipping HTTP Request, too many failures have occurred.\nLast error: {0}")]

crates/turborepo-api-client/src/lib.rs

@@ -8,7 +8,7 @@ use std::{backtrace::Backtrace, env, future::Future, time::Duration};
 use lazy_static::lazy_static;
 use regex::Regex;
 pub use reqwest::Response;
-use reqwest::{Method, RequestBuilder, StatusCode};
+use reqwest::{Body, Method, RequestBuilder, StatusCode};
 use serde::Deserialize;
 use turborepo_ci::{is_ci, Vendor};
 use turborepo_vercel_api::{
@@ -26,6 +26,9 @@ mod retry;
 pub mod spaces;
 pub mod telemetry;
 
+pub use bytes::Bytes;
+pub use tokio_stream::Stream;
+
 lazy_static! {
     static ref AUTHORIZATION_REGEX: Regex =
         Regex::new(r"(?i)(?:^|,) *authorization *(?:,|$)").unwrap();
@@ -74,7 +77,7 @@ pub trait CacheClient {
     fn put_artifact(
         &self,
         hash: &str,
-        artifact_body: &[u8],
+        artifact_body: impl tokio_stream::Stream<Item = Result<bytes::Bytes>> + Send + Sync + 'static,
         duration: u64,
         tag: Option<&str>,
         token: &str,
@@ -358,7 +361,7 @@ impl CacheClient for APIClient {
     async fn put_artifact(
         &self,
         hash: &str,
-        artifact_body: &[u8],
+        artifact_body: impl tokio_stream::Stream<Item = Result<bytes::Bytes>> + Send + Sync + 'static,
         duration: u64,
         tag: Option<&str>,
         token: &str,
@@ -382,13 +385,15 @@ impl CacheClient for APIClient {
             request_url = preflight_response.location.clone();
         }
 
+        let stream = Body::wrap_stream(artifact_body);
+
         let mut request_builder = self
             .cache_client
             .put(request_url)
             .header("Content-Type", "application/octet-stream")
             .header("x-artifact-duration", duration.to_string())
             .header("User-Agent", self.user_agent.clone())
-            .body(artifact_body.to_vec());
+            .body(stream);
 
         if allow_auth {
             request_builder = request_builder.header("Authorization", format!("Bearer {}", token));

crates/turborepo-auth/src/auth/login.rs

@@ -313,7 +313,11 @@ mod tests {
         async fn put_artifact(
             &self,
             _hash: &str,
-            _artifact_body: &[u8],
+            _artifact_body: impl turborepo_api_client::Stream<
+                    Item = Result<turborepo_api_client::Bytes, turborepo_api_client::Error>,
+                > + Send
+                + Sync
+                + 'static,
             _duration: u64,
             _tag: Option<&str>,
             _token: &str,

crates/turborepo-auth/src/auth/sso.rs

@@ -310,7 +310,11 @@ mod tests {
         async fn put_artifact(
             &self,
             _hash: &str,
-            _artifact_body: &[u8],
+            _artifact_body: impl turborepo_api_client::Stream<
+                    Item = Result<turborepo_api_client::Bytes, turborepo_api_client::Error>,
+                > + Send
+                + Sync
+                + 'static,
             _duration: u64,
             _tag: Option<&str>,
             _token: &str,

crates/turborepo-auth/src/lib.rs

@@ -420,7 +420,11 @@ mod tests {
         async fn put_artifact(
             &self,
             _hash: &str,
-            _artifact_body: &[u8],
+            _artifact_body: impl turborepo_api_client::Stream<
+                    Item = Result<turborepo_api_client::Bytes, turborepo_api_client::Error>,
+                > + Send
+                + Sync
+                + 'static,
             _duration: u64,
             _tag: Option<&str>,
             _token: &str,