Skip to content

Commit

Permalink
add support for cache_timeout as well as timeout (#8078)
Browse files Browse the repository at this point in the history
### Description

We currently have a single API client however some types of calls should
be handled differently that others. The cache in particular should not
cause a hard error if it times out during upload, only if it fails to
connect. `cache_client` is currently only used for uploading artifacts.

TODO: decide on an env var name and add docs. Going to defer to monday's
meeting for this.

### Testing Instructions

New unit tests.

Closes TURBO-2974
  • Loading branch information
arlyon committed May 8, 2024
1 parent 52552eb commit 16068cb
Show file tree
Hide file tree
Showing 7 changed files with 138 additions and 24 deletions.
60 changes: 47 additions & 13 deletions crates/turborepo-api-client/src/lib.rs
Expand Up @@ -2,7 +2,7 @@
#![feature(error_generic_member_access)]
#![deny(clippy::all)]

use std::{backtrace::Backtrace, env, future::Future};
use std::{backtrace::Backtrace, env, future::Future, time::Duration};

use lazy_static::lazy_static;
use regex::Regex;
Expand Down Expand Up @@ -106,6 +106,7 @@ pub trait TokenClient {
#[derive(Clone)]
pub struct APIClient {
client: reqwest::Client,
cache_client: reqwest::Client,
base_url: String,
user_agent: String,
use_preflight: bool,
Expand Down Expand Up @@ -371,7 +372,7 @@ impl CacheClient for APIClient {
}

let mut request_builder = self
.client
.cache_client
.put(request_url)
.header("Content-Type", "application/octet-stream")
.header("x-artifact-duration", duration.to_string())
Expand Down Expand Up @@ -534,25 +535,50 @@ impl TokenClient for APIClient {
}

impl APIClient {
/// Create a new APIClient.
///
/// # Arguments
/// `base_url` - The base URL for the API.
/// `timeout` - The timeout for requests.
/// `upload_timeout` - If specified, uploading files will use `timeout` for
/// the connection, and `upload_timeout` for the total.
/// Otherwise, `timeout` will be used for the total.
/// `version` - The version of the client.
/// `use_preflight` - If true, use the preflight API for all requests.
pub fn new(
base_url: impl AsRef<str>,
timeout: u64,
timeout: Option<Duration>,
upload_timeout: Option<Duration>,
version: &str,
use_preflight: bool,
) -> Result<Self> {
let client_build = if timeout != 0 {
reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(timeout))
.build()
// for the api client, the timeout applies for the entire duration
// of the request, including the connection phase
let client = reqwest::Client::builder();
let client = if let Some(dur) = timeout {
client.timeout(dur)
} else {
reqwest::Client::builder().build()
};

let client = client_build.map_err(Error::TlsError)?;
client
}
.build()
.map_err(Error::TlsError)?;

// for the cache client, the timeout applies only to the request
// connection time, while the upload timeout applies to the entire
// request
let cache_client = reqwest::Client::builder();
let cache_client = match (timeout, upload_timeout) {
(Some(dur), Some(upload_dur)) => cache_client.connect_timeout(dur).timeout(upload_dur),
(Some(dur), None) | (None, Some(dur)) => cache_client.timeout(dur),
(None, None) => cache_client,
}
.build()
.map_err(Error::TlsError)?;

let user_agent = build_user_agent(version);
Ok(APIClient {
client,
cache_client,
base_url: base_url.as_ref().to_string(),
user_agent,
use_preflight,
Expand Down Expand Up @@ -708,7 +734,7 @@ impl AnonAPIClient {
pub fn new(base_url: impl AsRef<str>, timeout: u64, version: &str) -> Result<Self> {
let client_build = if timeout != 0 {
reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(timeout))
.timeout(Duration::from_secs(timeout))
.build()
} else {
reqwest::Client::builder().build()
Expand Down Expand Up @@ -737,6 +763,8 @@ fn build_user_agent(version: &str) -> String {

#[cfg(test)]
mod test {
use std::time::Duration;

use anyhow::Result;
use turborepo_vercel_api_mock::start_test_server;
use url::Url;
Expand All @@ -749,7 +777,13 @@ mod test {
let handle = tokio::spawn(start_test_server(port));
let base_url = format!("http://localhost:{}", port);

let client = APIClient::new(&base_url, 200, "2.0.0", true)?;
let client = APIClient::new(
&base_url,
Some(Duration::from_secs(200)),
None,
"2.0.0",
true,
)?;

let response = client
.do_preflight(
Expand Down
26 changes: 22 additions & 4 deletions crates/turborepo-cache/src/async_cache.rs
Expand Up @@ -186,7 +186,7 @@ impl AsyncCache {

#[cfg(test)]
mod tests {
use std::assert_matches::assert_matches;
use std::{assert_matches::assert_matches, time::Duration};

use anyhow::Result;
use futures::future::try_join_all;
Expand Down Expand Up @@ -235,7 +235,13 @@ mod tests {
}),
};

let api_client = APIClient::new(format!("http://localhost:{}", port), 200, "2.0.0", true)?;
let api_client = APIClient::new(
format!("http://localhost:{}", port),
Some(Duration::from_secs(200)),
None,
"2.0.0",
true,
)?;
let api_auth = Some(APIAuth {
team_id: Some("my-team-id".to_string()),
token: "my-token".to_string(),
Expand Down Expand Up @@ -317,7 +323,13 @@ mod tests {

// Initialize client with invalid API url to ensure that we don't hit the
// network
let api_client = APIClient::new("http://example.com", 200, "2.0.0", true)?;
let api_client = APIClient::new(
"http://example.com",
Some(Duration::from_secs(200)),
None,
"2.0.0",
true,
)?;
let api_auth = Some(APIAuth {
team_id: Some("my-team-id".to_string()),
token: "my-token".to_string(),
Expand Down Expand Up @@ -405,7 +417,13 @@ mod tests {
}),
};

let api_client = APIClient::new(format!("http://localhost:{}", port), 200, "2.0.0", true)?;
let api_client = APIClient::new(
format!("http://localhost:{}", port),
Some(Duration::from_secs(200)),
None,
"2.0.0",
true,
)?;
let api_auth = Some(APIAuth {
team_id: Some("my-team-id".to_string()),
token: "my-token".to_string(),
Expand Down
10 changes: 9 additions & 1 deletion crates/turborepo-cache/src/fs.rs
Expand Up @@ -182,6 +182,8 @@ impl FSCache {

#[cfg(test)]
mod test {
use std::time::Duration;

use anyhow::Result;
use futures::future::try_join_all;
use tempfile::tempdir;
Expand Down Expand Up @@ -216,7 +218,13 @@ mod test {
let repo_root_path = AbsoluteSystemPath::from_std_path(repo_root.path())?;
test_case.initialize(repo_root_path)?;

let api_client = APIClient::new(format!("http://localhost:{}", port), 200, "2.0.0", true)?;
let api_client = APIClient::new(
format!("http://localhost:{}", port),
Some(Duration::from_secs(200)),
None,
"2.0.0",
true,
)?;
let api_auth = APIAuth {
team_id: Some("my-team".to_string()),
token: "my-token".to_string(),
Expand Down
10 changes: 9 additions & 1 deletion crates/turborepo-cache/src/http.rs
Expand Up @@ -235,6 +235,8 @@ impl HTTPCache {

#[cfg(test)]
mod test {
use std::time::Duration;

use anyhow::Result;
use futures::future::try_join_all;
use tempfile::tempdir;
Expand Down Expand Up @@ -276,7 +278,13 @@ mod test {
let files = &test_case.files;
let duration = test_case.duration;

let api_client = APIClient::new(format!("http://localhost:{}", port), 200, "2.0.0", true)?;
let api_client = APIClient::new(
format!("http://localhost:{}", port),
Some(Duration::from_secs(200)),
None,
"2.0.0",
true,
)?;
let opts = CacheOpts::default();
let api_auth = APIAuth {
team_id: Some("my-team".to_string()),
Expand Down
23 changes: 19 additions & 4 deletions crates/turborepo-lib/src/commands/mod.rs
@@ -1,4 +1,4 @@
use std::cell::OnceCell;
use std::{cell::OnceCell, time::Duration};

use turbopath::{AbsoluteSystemPath, AbsoluteSystemPathBuf};
use turborepo_api_client::{APIAuth, APIClient};
Expand Down Expand Up @@ -125,9 +125,24 @@ impl CommandBase {
let config = self.config()?;
let api_url = config.api_url();
let timeout = config.timeout();

APIClient::new(api_url, timeout, self.version, config.preflight())
.map_err(ConfigError::ApiClient)
let upload_timeout = config.upload_timeout();

APIClient::new(
api_url,
if timeout > 0 {
Some(Duration::from_secs(timeout))
} else {
None
},
if upload_timeout > 0 {
Some(Duration::from_secs(upload_timeout))
} else {
None
},
self.version,
config.preflight(),
)
.map_err(ConfigError::ApiClient)
}

/// Current working directory for the turbo command
Expand Down
23 changes: 23 additions & 0 deletions crates/turborepo-lib/src/config.rs
Expand Up @@ -135,6 +135,8 @@ pub enum Error {
InvalidRemoteCacheEnabled,
#[error("TURBO_REMOTE_CACHE_TIMEOUT: error parsing timeout.")]
InvalidRemoteCacheTimeout(#[source] std::num::ParseIntError),
#[error("TURBO_REMOTE_CACHE_UPLOAD_TIMEOUT: error parsing timeout.")]
InvalidUploadTimeout(#[source] std::num::ParseIntError),
#[error("TURBO_PREFLIGHT should be either 1 or 0.")]
InvalidPreflight,
#[error(transparent)]
Expand All @@ -154,6 +156,7 @@ macro_rules! create_builder {
const DEFAULT_API_URL: &str = "https://vercel.com/api";
const DEFAULT_LOGIN_URL: &str = "https://vercel.com";
const DEFAULT_TIMEOUT: u64 = 30;
const DEFAULT_UPLOAD_TIMEOUT: u64 = 60;

// We intentionally don't derive Serialize so that different parts
// of the code that want to display the config can tune how they
Expand Down Expand Up @@ -181,6 +184,7 @@ pub struct ConfigurationOptions {
pub(crate) signature: Option<bool>,
pub(crate) preflight: Option<bool>,
pub(crate) timeout: Option<u64>,
pub(crate) upload_timeout: Option<u64>,
pub(crate) enabled: Option<bool>,
pub(crate) spaces_id: Option<String>,
#[serde(rename = "experimentalUI")]
Expand Down Expand Up @@ -234,10 +238,16 @@ impl ConfigurationOptions {
self.preflight.unwrap_or_default()
}

/// Note: 0 implies no timeout
pub fn timeout(&self) -> u64 {
self.timeout.unwrap_or(DEFAULT_TIMEOUT)
}

/// Note: 0 implies no timeout
pub fn upload_timeout(&self) -> u64 {
self.upload_timeout.unwrap_or(DEFAULT_UPLOAD_TIMEOUT)
}

pub fn spaces_id(&self) -> Option<&str> {
self.spaces_id.as_deref()
}
Expand Down Expand Up @@ -312,6 +322,7 @@ fn get_env_var_config(
turbo_mapping.insert(OsString::from("turbo_teamid"), "team_id");
turbo_mapping.insert(OsString::from("turbo_token"), "token");
turbo_mapping.insert(OsString::from("turbo_remote_cache_timeout"), "timeout");
turbo_mapping.insert(OsString::from("turbo_api_timeout"), "api_timeout");
turbo_mapping.insert(OsString::from("turbo_experimental_ui"), "experimental_ui");
turbo_mapping.insert(OsString::from("turbo_preflight"), "preflight");

Expand Down Expand Up @@ -383,6 +394,16 @@ fn get_env_var_config(
None
};

let upload_timeout = if let Some(upload_timeout) = output_map.get("upload_timeout") {
Some(
upload_timeout
.parse::<u64>()
.map_err(Error::InvalidUploadTimeout)?,
)
} else {
None
};

// Process experimentalUI
let experimental_ui = output_map
.get("experimental_ui")
Expand Down Expand Up @@ -412,6 +433,7 @@ fn get_env_var_config(

// Processed numbers
timeout,
upload_timeout,
spaces_id,
};

Expand Down Expand Up @@ -457,6 +479,7 @@ fn get_override_env_var_config(
enabled: None,
experimental_ui: None,
timeout: None,
upload_timeout: None,
spaces_id: None,
};

Expand Down
10 changes: 9 additions & 1 deletion crates/turborepo-lib/src/run/summary/spaces.rs
Expand Up @@ -351,6 +351,8 @@ fn trim_logs(logs: &[u8], limit: usize) -> String {

#[cfg(test)]
mod tests {
use std::time::Duration;

use anyhow::Result;
use chrono::Local;
use pretty_assertions::assert_eq;
Expand All @@ -375,7 +377,13 @@ mod tests {
let port = port_scanner::request_open_port().unwrap();
let handle = tokio::spawn(start_test_server(port));

let api_client = APIClient::new(format!("http://localhost:{}", port), 2, "", true)?;
let api_client = APIClient::new(
format!("http://localhost:{}", port),
Some(Duration::from_secs(2)),
None,
"",
true,
)?;

let api_auth = Some(APIAuth {
token: EXPECTED_TOKEN.to_string(),
Expand Down

0 comments on commit 16068cb

Please sign in to comment.