diff --git a/Cargo.lock b/Cargo.lock index 5faddfe67..f87359a05 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3410,6 +3410,7 @@ dependencies = [ "globset", "hex", "hmac", + "httpdate", "ignore", "indicatif 0.17.11", "insta", diff --git a/Cargo.toml b/Cargo.toml index 360ade344..03f6e4991 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -89,6 +89,7 @@ tokio-util = { version = "0.7", features = ["io", "io-util"] } # HTTP reqwest = { version = "0.12", features = ["json", "rustls-tls", "stream", "blocking", "http2", "gzip", "brotli", "zstd"], default-features = false } +httpdate = "1" axum = { version = "0.8", features = ["json"] } # Serialization diff --git a/crates/lpm-cli/Cargo.toml b/crates/lpm-cli/Cargo.toml index 2a5a738f3..87732ce0a 100644 --- a/crates/lpm-cli/Cargo.toml +++ b/crates/lpm-cli/Cargo.toml @@ -58,6 +58,7 @@ cvss = { workspace = true } tokio = { workspace = true } tokio-util = { workspace = true, features = ["rt"] } reqwest = { workspace = true } +httpdate = { workspace = true } axum = { workspace = true } serde = { workspace = true, features = ["rc"] } serde_json = { workspace = true } diff --git a/crates/lpm-cli/src/commands/login.rs b/crates/lpm-cli/src/commands/login.rs index 2cc1517eb..ced8626dc 100644 --- a/crates/lpm-cli/src/commands/login.rs +++ b/crates/lpm-cli/src/commands/login.rs @@ -189,33 +189,14 @@ pub async fn run( } let exchange_url = format!("{registry_url}/api/cli/exchange"); let http_client = exchange_client(std::time::Duration::from_secs(30))?; - let resp = http_client - .post(&exchange_url) - .json(&serde_json::json!({ "code": code, "code_verifier": code_verifier })) - .send() - .await - .map_err(|error| { - LpmError::Registry(format!( - "exchange request failed: {}", - lpm_http::display_error(&error) - )) - })?; - - let status = resp.status(); - let body = read_capped_exchange_body(resp).await?; - if !status.is_success() { - let detail = if let Ok(json) = serde_json::from_slice::(&body) { - json["error"] - .as_str() - .unwrap_or("unknown error") - .to_string() - } else { - format!("HTTP {status}") - }; - return Err(LpmError::Registry(format!( - "Failed to exchange authorization code: {detail}. It may have expired — please try again." - ))); - } + let body = exchange_authorization_code( + &http_client, + &exchange_url, + &code, + &code_verifier, + EXCHANGE_RETRY_BUDGET, + ) + .await?; let session = parse_cli_exchange_session(&body)?; let token = session.token; @@ -251,6 +232,84 @@ pub async fn run( Ok(()) } +/// How long `lpm login` keeps asking the Registry to redeem an authorization +/// code it answered "try again" for, as while it settles a code's state. +const EXCHANGE_RETRY_BUDGET: std::time::Duration = std::time::Duration::from_secs(30); +/// The longest wait between attempts, whatever `Retry-After` asks. +const EXCHANGE_RETRY_MAX_WAIT: std::time::Duration = std::time::Duration::from_secs(5); + +/// Redeems the authorization code for a session, and returns the response +/// body. The Registry answers 503 while it can't yet tell whether the code +/// was redeemed, and 429 when the address is rate-limited; both keep the +/// code, so they're asked again after their `Retry-After` while `budget` +/// allows. Any other refusal is final, as the code is spent or invalid. +async fn exchange_authorization_code( + client: &reqwest::Client, + exchange_url: &str, + code: &str, + code_verifier: &str, + budget: std::time::Duration, +) -> Result, LpmError> { + let started = std::time::Instant::now(); + loop { + let response = client + .post(exchange_url) + .json(&serde_json::json!({ "code": code, "code_verifier": code_verifier })) + .send() + .await + .map_err(|error| { + LpmError::Registry(format!( + "exchange request failed: {}", + lpm_http::display_error(&error) + )) + })?; + let status = response.status(); + let retryable = status == reqwest::StatusCode::SERVICE_UNAVAILABLE + || status == reqwest::StatusCode::TOO_MANY_REQUESTS; + let wait = exchange_retry_wait(&response); + let body = read_capped_exchange_body(response).await?; + if status.is_success() { + return Ok(body); + } + if retryable && started.elapsed() + wait <= budget { + tokio::time::sleep(wait).await; + continue; + } + let detail = serde_json::from_slice::(&body) + .ok() + .and_then(|json| json["error"].as_str().map(str::to_string)) + .unwrap_or_else(|| format!("HTTP {}", status.as_u16())); + return Err(LpmError::Registry(if retryable { + format!("Failed to exchange authorization code: {detail}. Run `lpm login` again.") + } else { + format!( + "Failed to exchange authorization code: {detail}. It may have expired — please try again." + ) + })); + } +} + +/// The wait a response's `Retry-After` asks, in whole seconds, or a second +/// when it names none, at most `EXCHANGE_RETRY_MAX_WAIT`. +fn exchange_retry_wait(response: &reqwest::Response) -> std::time::Duration { + retry_after_wait( + response + .headers() + .get(reqwest::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()), + ) +} + +fn retry_after_wait(retry_after: Option<&str>) -> std::time::Duration { + retry_after + .and_then(|value| value.trim().parse::().ok()) + .map_or( + std::time::Duration::from_secs(1), + std::time::Duration::from_secs, + ) + .min(EXCHANGE_RETRY_MAX_WAIT) +} + fn exchange_client(timeout: std::time::Duration) -> Result { lpm_http::client_builder() .redirect(reqwest::redirect::Policy::none()) @@ -757,6 +816,142 @@ mod tests { } } + async fn exchange_server(responses: Vec) -> wiremock::MockServer { + use wiremock::{Mock, MockServer}; + let server = MockServer::start().await; + for (index, response) in responses.into_iter().enumerate() { + Mock::given(wiremock::matchers::method("POST")) + .respond_with(response) + .up_to_n_times(1) + .with_priority(u8::try_from(index + 1).unwrap()) + .mount(&server) + .await; + } + server + } + + #[tokio::test] + async fn exchange_retries_while_the_registry_settles_the_code_then_succeeds() { + use wiremock::ResponseTemplate; + let unsettled = || { + ResponseTemplate::new(503) + .insert_header("retry-after", "0") + .set_body_json(serde_json::json!({"error": "The sign-in can't be completed right now; try again"})) + }; + let server = exchange_server(vec![ + unsettled(), + ResponseTemplate::new(429).insert_header("retry-after", "0"), + unsettled(), + ResponseTemplate::new(200).set_body_json(valid_exchange_session()), + ]) + .await; + let client = exchange_client(std::time::Duration::from_secs(5)).unwrap(); + + let body = exchange_authorization_code( + &client, + &server.uri(), + "the-code", + "the-verifier", + std::time::Duration::from_secs(10), + ) + .await + .unwrap(); + + assert!(parse_cli_exchange_session(&body).is_ok()); + let requests = server.received_requests().await.unwrap(); + assert_eq!(requests.len(), 4); + for request in requests { + let sent: serde_json::Value = serde_json::from_slice(&request.body).unwrap(); + assert_eq!( + sent, + serde_json::json!({"code": "the-code", "code_verifier": "the-verifier"}) + ); + } + } + + #[tokio::test] + async fn exchange_stops_retrying_at_its_budget_with_the_registry_message() { + use wiremock::{Mock, MockServer, ResponseTemplate}; + let server = MockServer::start().await; + Mock::given(wiremock::matchers::method("POST")) + .respond_with( + ResponseTemplate::new(503) + .insert_header("retry-after", "1") + .set_body_json(serde_json::json!({"error": "The sign-in can't be completed right now; try again"})), + ) + .mount(&server) + .await; + let client = exchange_client(std::time::Duration::from_secs(5)).unwrap(); + let started = std::time::Instant::now(); + + let error = exchange_authorization_code( + &client, + &server.uri(), + "the-code", + "the-verifier", + std::time::Duration::from_millis(1500), + ) + .await + .unwrap_err(); + + assert!(started.elapsed() < std::time::Duration::from_secs(3)); + assert!( + error + .to_string() + .contains("The sign-in can't be completed right now"), + "{error}" + ); + assert!( + error.to_string().contains("Run `lpm login` again."), + "{error}" + ); + assert!(!error.to_string().contains("expired"), "{error}"); + assert_eq!(server.received_requests().await.unwrap().len(), 2); + } + + #[test] + fn exchange_waits_what_retry_after_asks_within_its_cap() { + let second = std::time::Duration::from_secs(1); + assert_eq!(retry_after_wait(Some("0")), std::time::Duration::ZERO); + assert_eq!(retry_after_wait(Some(" 2 ")), 2 * second); + assert_eq!(retry_after_wait(Some("3600")), EXCHANGE_RETRY_MAX_WAIT); + assert_eq!( + retry_after_wait(Some("Wed, 21 Oct 2026 07:28:00 GMT")), + second + ); + assert_eq!(retry_after_wait(None), second); + } + + #[tokio::test] + async fn exchange_does_not_retry_a_refused_code() { + use wiremock::ResponseTemplate; + let server = exchange_server(vec![ + ResponseTemplate::new(401) + .set_body_json(serde_json::json!({"error": "Invalid or expired exchange code"})), + ResponseTemplate::new(200).set_body_json(valid_exchange_session()), + ]) + .await; + let client = exchange_client(std::time::Duration::from_secs(5)).unwrap(); + + let error = exchange_authorization_code( + &client, + &server.uri(), + "the-code", + "the-verifier", + std::time::Duration::from_secs(10), + ) + .await + .unwrap_err(); + + assert!( + error + .to_string() + .contains("Invalid or expired exchange code"), + "{error}" + ); + assert_eq!(server.received_requests().await.unwrap().len(), 1); + } + #[test] fn verification_preserves_explicit_insecure_transport_choice() { let client = RegistryClient::new().with_insecure(true); diff --git a/crates/lpm-cli/src/commands/remote_cache.rs b/crates/lpm-cli/src/commands/remote_cache.rs index 07cf86d49..c97f0a687 100644 --- a/crates/lpm-cli/src/commands/remote_cache.rs +++ b/crates/lpm-cli/src/commands/remote_cache.rs @@ -4,22 +4,34 @@ use base64::engine::general_purpose::{STANDARD_NO_PAD, URL_SAFE_NO_PAD}; use hmac::{Hmac, Mac}; use lpm_common::{LpmError, LpmRoot, format_bytes}; use reqwest::StatusCode; -use reqwest::header::CONTENT_LENGTH; +use reqwest::header::{CONTENT_LENGTH, LOCATION}; use reqwest::{Body, Client}; use sha2::{Digest, Sha256}; -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::fs::File; use std::future::Future; +use std::hash::{BuildHasher, RandomState}; use std::io::{IsTerminal, Read, Seek, Write}; use std::path::{Path, PathBuf}; -use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::time::Duration; +use std::sync::{Arc, LazyLock, Mutex, PoisonError}; +use std::time::{Duration, Instant, SystemTime}; type HmacSha256 = Hmac; +/// The largest artifact the LPM.dev Registry stores. +const REGISTRY_MAX_ARTIFACT_BYTES: u64 = 100 * 1024 * 1024; +/// The largest artifact uploaded to, or restored from, any other cache server. const MAX_REMOTE_ARTIFACT_BYTES: u64 = 500 * 1024 * 1024; const MAX_REMOTE_STATUS_BYTES: u64 = 64 * 1024; +/// How long a 429 pauses requests when it gives no `Retry-After`: a minute, +/// the Registry's shortest window. +const DEFAULT_LIMIT_PAUSE: Duration = Duration::from_secs(60); +/// The longest pause a 429 can ask for: a day, the Registry's longest window. +const MAX_LIMIT_PAUSE: Duration = Duration::from_secs(24 * 60 * 60); +/// The longest wait an upload refused with 429 sits out before going once +/// more; a longer one is a spent limit. +const MAX_RETRY_WAIT: Duration = Duration::from_secs(5); const REMOTE_CACHE_TIMEOUT: Duration = Duration::from_secs(30); /// Each upload read moves to the blocking pool, so the stream's 4 KiB default /// made a 64 MiB upload about ten times slower than 64 KiB reads. @@ -29,8 +41,17 @@ const ARTIFACT_SHA_HEADER: &str = "x-artifact-sha"; const ARTIFACT_DURATION_HEADER: &str = "x-artifact-duration"; const ARTIFACT_CLIENT_CI_HEADER: &str = "x-artifact-client-ci"; const ARTIFACT_CLIENT_INTERACTIVE_HEADER: &str = "x-artifact-client-interactive"; +/// Asks the LPM.dev Registry to answer a cache hit with a redirect to the +/// cache's download host, the artifact's headers on the redirect, rather +/// than streaming it. Other cache servers ignore it. +const DOWNLOAD_REDIRECT_HEADER: &str = "x-lpm-download-redirect"; static REMOTE_CACHE_WARNING_SHOWN: AtomicBool = AtomicBool::new(false); +/// A cache server's base URL and the team its requests name. +type CacheTeam = (String, Option); +/// Each cache server and team's request limits, shared by every task's client +/// in the run. +static REQUEST_LIMITS: Mutex>> = Mutex::new(BTreeMap::new()); #[derive(Clone)] pub struct RemoteCacheClient { @@ -40,6 +61,156 @@ pub struct RemoteCacheClient { signature_key: Option, read_only: bool, env_policy: lpm_runner::lpm_json::RemoteCacheEnvConfig, + max_artifact_bytes: u64, + limits: Arc, +} + +#[derive(Clone, Copy)] +enum LimitedRequest { + Lookup, + Upload, +} + +/// Until when a cache server refuses one team's lookups, and its uploads, +/// after answering one with 429, until when of that for a spent limit, and +/// whether the run has said so yet. +#[derive(Default)] +struct RequestLimits { + lookups_until: Mutex>, + uploads_until: Mutex>, + lookups_limited_until: Mutex>, + uploads_limited_until: Mutex>, + lookup_limit_warned: AtomicBool, + upload_limit_warned: AtomicBool, + lookups_busy_noted: AtomicBool, + uploads_busy_noted: AtomicBool, + uploads_refused: AtomicBool, + lookups_refused: AtomicBool, +} + +impl RequestLimits { + fn for_cache(base_url: &str, team: Option<&str>) -> Arc { + REQUEST_LIMITS + .lock() + .unwrap_or_else(PoisonError::into_inner) + .entry((base_url.to_owned(), team.map(ToOwned::to_owned))) + .or_default() + .clone() + } + + fn until(&self, request: LimitedRequest) -> &Mutex> { + match request { + LimitedRequest::Lookup => &self.lookups_until, + LimitedRequest::Upload => &self.uploads_until, + } + } + + fn limited_until(&self, request: LimitedRequest) -> &Mutex> { + match request { + LimitedRequest::Lookup => &self.lookups_limited_until, + LimitedRequest::Upload => &self.uploads_limited_until, + } + } + + fn paused(&self, request: LimitedRequest) -> bool { + running(self.until(request)) + } + + fn lookups_paused(&self) -> bool { + self.paused(LimitedRequest::Lookup) + } + + /// Uploads also wait out a lookup limit: a task that couldn't look its + /// artifact up would likely upload one the cache already holds. + fn uploads_paused(&self) -> bool { + self.paused(LimitedRequest::Upload) || self.paused(LimitedRequest::Lookup) + } + + /// Whether this kind of request waits out a spent limit, rather than the + /// cache server's room for requests at once. + fn paused_by_a_limit(&self, request: LimitedRequest) -> bool { + running(self.limited_until(request)) + || (matches!(request, LimitedRequest::Upload) + && running(self.limited_until(LimitedRequest::Lookup))) + } + + /// Pauses this kind of request while the cache server has no room for it. + fn pause(&self, request: LimitedRequest, wait: Duration) { + extend(self.until(request), Instant::now() + wait); + } + + /// Pauses this kind of request for a spent limit, whose warning covers + /// the requests it skips. + fn spend(&self, request: LimitedRequest, wait: Duration) { + let resume = Instant::now() + wait; + extend(self.until(request), resume); + extend(self.limited_until(request), resume); + } + + /// Stops uploads for the rest of the run once the cache server refuses + /// one: the token can't write there, and every later upload would be + /// refused too, each counting against the server's limit for refused + /// requests. True when this call stopped them, so the run says so once. + fn refuse_uploads(&self) -> bool { + !self.uploads_refused.swap(true, Ordering::Relaxed) + } + + fn uploads_refused(&self) -> bool { + self.uploads_refused.load(Ordering::Relaxed) + } + + /// Stops lookups and uploads for the rest of the run once the cache + /// server refuses a lookup, or bills nothing more: the token can't read + /// that cache, or the cache is paused or disabled, so every later + /// request would be refused too. True when this call stopped lookups. + fn refuse_lookups(&self) -> bool { + self.uploads_refused.store(true, Ordering::Relaxed); + !self.lookups_refused.swap(true, Ordering::Relaxed) + } + + fn lookups_refused(&self) -> bool { + self.lookups_refused.load(Ordering::Relaxed) + } + + /// True the first time the run reaches this kind of limit. + fn first_warning(&self, request: LimitedRequest) -> bool { + let warned = match request { + LimitedRequest::Lookup => &self.lookup_limit_warned, + LimitedRequest::Upload => &self.upload_limit_warned, + }; + !warned.swap(true, Ordering::Relaxed) + } + + /// True the first time the run skips this kind of request because the + /// cache server was busy for a moment. + fn first_busy_notice(&self, request: LimitedRequest) -> bool { + let noted = match request { + LimitedRequest::Lookup => &self.lookups_busy_noted, + LimitedRequest::Upload => &self.uploads_busy_noted, + }; + !noted.swap(true, Ordering::Relaxed) + } +} + +fn running(until: &Mutex>) -> bool { + until + .lock() + .unwrap_or_else(PoisonError::into_inner) + .is_some_and(|until| Instant::now() < until) +} + +/// Moves `until` to `resume`, unless it's already later. +fn extend(until: &Mutex>, resume: Instant) { + let mut until = until.lock().unwrap_or_else(PoisonError::into_inner); + if until.is_none_or(|until| until < resume) { + *until = Some(resume); + } +} + +/// A 429's message, and how long to wait before the next request of its kind. +struct LimitReached { + message: String, + wait: Duration, } #[derive(Clone)] @@ -89,6 +260,13 @@ pub fn try_restore( output_globs: &[String], validate: impl FnOnce() -> Result, ) -> Option { + if client.limits.lookups_refused() { + return None; + } + if client.limits.lookups_paused() { + client.skipping_while_busy(LimitedRequest::Lookup); + return None; + } match client.restore(key, project_dir, output_globs, validate) { Ok(hit) => hit, Err(reason) => { @@ -111,7 +289,12 @@ pub fn try_store( env_vars: &HashMap, inherited_env: &HashMap, ) { - if client.read_only { + if client.read_only || client.limits.uploads_refused() { + return; + } + // A lookup's pause skips uploads too, so its notice alone wouldn't say so. + if client.limits.uploads_paused() { + client.skipping_while_busy(LimitedRequest::Upload); return; } @@ -375,79 +558,117 @@ impl RemoteCacheClient { output_globs: &[String], validate: impl FnOnce() -> Result, ) -> Result, String> { - let client = http_client()?; + let client = lookup_client()?; let url = self.artifact_url(key)?; - let mut response = self - .send_authenticated( - "remote cache authorization failed; continuing without it", - async |bearer: String| { - let response = client - .get(url.clone()) - .bearer_auth(bearer) - .send() - .await - .map_err(|error| { - RemoteCacheRequestError::Message(format!( - "remote cache lookup failed: {}", - lpm_http::display_error(&error) - )) - })?; - if response.status() == StatusCode::UNAUTHORIZED { - Err(RemoteCacheRequestError::Unauthorized) - } else { - Ok(response) - } - }, - ) - .await?; + let mut retried = false; + let response = loop { + let response = self + .send_authenticated( + "remote cache authorization failed; continuing without it", + async |bearer: String| { + let response = client + .get(url.clone()) + .bearer_auth(bearer) + .header(DOWNLOAD_REDIRECT_HEADER, "1") + .send() + .await + .map_err(|error| { + RemoteCacheRequestError::Message(format!( + "remote cache lookup failed: {}", + lpm_http::display_error(&error) + )) + })?; + if response.status() == StatusCode::UNAUTHORIZED { + Err(RemoteCacheRequestError::Unauthorized) + } else { + Ok(response) + } + }, + ) + .await?; + if response.status() != StatusCode::TOO_MANY_REQUESTS { + break response; + } + let limit = read_limit_reached(response).await; + // A wait this short is the server's room for requests at once, not + // a spent limit: a hit is worth looking up again after it, unless + // lookups were paused meanwhile. + if !retried && limit.wait <= MAX_RETRY_WAIT { + retried = true; + tokio::time::sleep(limit.wait + retry_jitter(key, limit.wait)).await; + if self.limits.lookups_refused() { + return Ok(None); + } + if self.limits.lookups_paused() { + self.skipping_while_busy(LimitedRequest::Lookup); + return Ok(None); + } + continue; + } + self.limit_reached(LimitedRequest::Lookup, limit); + return Ok(None); + }; - match response.status() { - StatusCode::OK => {} + let tag = artifact_header(&response, ARTIFACT_TAG_HEADER); + let advertised_sha = artifact_header(&response, ARTIFACT_SHA_HEADER); + let mut response = match response.status() { + StatusCode::OK => response, + status if status.is_redirection() => { + // The redirect carries what verifies the download; without it + // storage isn't asked, as the artifact couldn't be restored. + if advertised_sha.is_none() { + return Err(MISSING_CONTENT_HASH.into()); + } + if self.signature_key.is_some() && tag.is_none() { + return Err(MISSING_SIGNATURE_TAG.into()); + } + match fetch_redirected_artifact(&url, &response).await? { + Some(artifact) => artifact, + None => return Ok(None), + } + } StatusCode::NOT_FOUND => return Ok(None), + StatusCode::FORBIDDEN | StatusCode::PAYMENT_REQUIRED => { + let refusal = read_refusal(response, "lookup").await; + if self.limits.refuse_lookups() { + install_ui::warn_untrusted(&format!( + "{}; skipping the remote cache for this run", + refusal.detail + )); + } + return Ok(None); + } _ => return Err(remote_cache_http_error(response, "lookup").await), - } + }; if let Some(length) = response.content_length() - && length > MAX_REMOTE_ARTIFACT_BYTES + && length > self.max_artifact_bytes { - return Err(format!( - "remote cache artifact exceeds the {} limit; treating it as a miss", - format_bytes(MAX_REMOTE_ARTIFACT_BYTES), - )); + return Err(oversized_download(self.max_artifact_bytes)); } - let tag = response - .headers() - .get(ARTIFACT_TAG_HEADER) - .and_then(|value| value.to_str().ok()) - .map(ToOwned::to_owned); - let advertised_sha = response - .headers() - .get(ARTIFACT_SHA_HEADER) - .and_then(|value| value.to_str().ok()) - .map(ToOwned::to_owned); let mut temp = tempfile::NamedTempFile::new() .map_err(|e| format!("failed to create remote cache temp file: {e}"))?; - let digests = - copy_response_to_temp(&mut response, &mut temp, self.signature_key.as_deref()).await?; + let digests = copy_response_to_temp( + &mut response, + &mut temp, + self.signature_key.as_deref(), + self.max_artifact_bytes, + ) + .await?; // A downloaded artifact is extracted into the project tree only after // the advertised content hash passes. When a signing key is configured, // the HMAC must pass too, so a cache host without the key can't forge // restoreable bytes. let Some(advertised_sha) = &advertised_sha else { - return Err( - "remote cache artifact is missing a content hash; treating it as a miss".into(), - ); + return Err(MISSING_CONTENT_HASH.into()); }; verify_artifact_sha(advertised_sha, &digests.sha256_hex)?; if let Some(key) = &self.signature_key { let Some(tag) = tag else { - return Err( - "remote cache artifact is missing a signature tag; treating it as a miss" - .into(), - ); + return Err(MISSING_SIGNATURE_TAG.into()); }; verify_artifact_tag(key, &tag, &digests.hmac_tag)?; } @@ -499,7 +720,6 @@ impl RemoteCacheClient { stderr: &str, duration_ms: u64, ) -> Result<(), String> { - let client = http_client()?; let mut artifact = tempfile::NamedTempFile::new() .map_err(|e| format!("failed to create remote cache artifact temp file: {e}"))?; let artifact_path = artifact.path().to_path_buf(); @@ -521,82 +741,181 @@ impl RemoteCacheClient { .metadata() .map_err(|e| format!("failed to stat remote cache artifact: {e}"))? .len(); - if artifact_len > MAX_REMOTE_ARTIFACT_BYTES { + self.check_upload_size(artifact_len)?; + self.upload_artifact(key, artifact.as_file(), artifact_len, duration_ms) + .await + } + + fn check_upload_size(&self, artifact_len: u64) -> Result<(), String> { + if artifact_len > self.max_artifact_bytes { return Err(format!( "remote cache artifact exceeds the {} upload limit; skipping upload", - format_bytes(MAX_REMOTE_ARTIFACT_BYTES), + format_cap(self.max_artifact_bytes), )); } + Ok(()) + } + /// Uploads the artifact in `artifact`, going once more after a 429 with a + /// short wait, and pausing uploads after one that asks for longer. + async fn upload_artifact( + &self, + key: &str, + artifact: &File, + artifact_len: u64, + duration_ms: u64, + ) -> Result<(), String> { + let client = http_client()?; let hash_file = artifact - .as_file() .try_clone() .map_err(|e| format!("failed to retain remote cache artifact for hashing: {e}"))?; let digests = hash_artifact(hash_file, self.signature_key.as_deref())?; let url = self.artifact_url(key)?; - let response = self - .send_authenticated( - "remote cache upload was not authorized; continuing without it", - async |bearer: String| { - let mut body_file = artifact.as_file().try_clone().map_err(|error| { - RemoteCacheRequestError::Message(format!( - "failed to retain remote cache artifact for upload: {error}" - )) - })?; - body_file.rewind().map_err(|error| { - RemoteCacheRequestError::Message(format!( - "failed to rewind remote cache artifact for upload: {error}" - )) - })?; - let mut request = client - .put(url.clone()) - .bearer_auth(bearer) - .header(CONTENT_LENGTH, artifact_len.to_string()) - .header(ARTIFACT_DURATION_HEADER, duration_ms.to_string()) - .header(ARTIFACT_SHA_HEADER, digests.sha256_hex.as_str()) - .header( - ARTIFACT_CLIENT_CI_HEADER, - if is_ci_environment() { "1" } else { "0" }, - ) - .header( - ARTIFACT_CLIENT_INTERACTIVE_HEADER, - if std::io::stdout().is_terminal() { - "1" - } else { - "0" - }, - ) - .body(Body::wrap_stream( - tokio_util::io::ReaderStream::with_capacity( - tokio::fs::File::from_std(body_file), - UPLOAD_READ_BYTES, - ), - )); - - if let Some(tag) = digests.hmac_tag.as_deref() { - request = request.header(ARTIFACT_TAG_HEADER, tag); - } - - let response = request.send().await.map_err(|error| { - RemoteCacheRequestError::Message(format!( - "remote cache upload failed: {}", - lpm_http::display_error(&error) - )) - })?; - if response.status() == StatusCode::UNAUTHORIZED { - Err(RemoteCacheRequestError::Unauthorized) - } else { - Ok(response) - } - }, - ) - .await?; + let mut retried = false; + loop { + let response = self + .send_authenticated( + "remote cache upload was not authorized; continuing without it", + async |bearer: String| { + let mut body_file = artifact.try_clone().map_err(|error| { + RemoteCacheRequestError::Message(format!( + "failed to retain remote cache artifact for upload: {error}" + )) + })?; + body_file.rewind().map_err(|error| { + RemoteCacheRequestError::Message(format!( + "failed to rewind remote cache artifact for upload: {error}" + )) + })?; + let mut request = client + .put(url.clone()) + .bearer_auth(bearer) + .header(CONTENT_LENGTH, artifact_len.to_string()) + .header(ARTIFACT_DURATION_HEADER, duration_ms.to_string()) + .header(ARTIFACT_SHA_HEADER, digests.sha256_hex.as_str()) + .header( + ARTIFACT_CLIENT_CI_HEADER, + if is_ci_environment() { "1" } else { "0" }, + ) + .header( + ARTIFACT_CLIENT_INTERACTIVE_HEADER, + if std::io::stdout().is_terminal() { + "1" + } else { + "0" + }, + ) + .body(Body::wrap_stream( + tokio_util::io::ReaderStream::with_capacity( + tokio::fs::File::from_std(body_file), + UPLOAD_READ_BYTES, + ), + )); + + if let Some(tag) = digests.hmac_tag.as_deref() { + request = request.header(ARTIFACT_TAG_HEADER, tag); + } + + let response = request.send().await.map_err(|error| { + RemoteCacheRequestError::Message(format!( + "remote cache upload failed: {}", + lpm_http::display_error(&error) + )) + })?; + if response.status() == StatusCode::UNAUTHORIZED { + Err(RemoteCacheRequestError::Unauthorized) + } else { + Ok(response) + } + }, + ) + .await?; - if response.status().is_success() { + // A conflict is another upload storing this artifact, which serves + // the next lookup as well as this one would. + if response.status().is_success() || response.status() == StatusCode::CONFLICT { + return Ok(()); + } + let status = response.status(); + if matches!(status, StatusCode::FORBIDDEN | StatusCode::PAYMENT_REQUIRED) { + let refusal = read_refusal(response, "upload").await; + // A cache too full for this upload may still take a later or + // smaller one. + if refusal.code.as_deref() == Some(QUOTA_EXCEEDED_CODE) { + return Err(format!("{}; continuing without it", refusal.detail)); + } + let (stopped, skipped) = if status == StatusCode::PAYMENT_REQUIRED { + (self.limits.refuse_lookups(), "the remote cache") + } else { + (self.limits.refuse_uploads(), "remote cache uploads") + }; + if stopped { + install_ui::warn_untrusted(&format!( + "{}; skipping {skipped} for this run", + refusal.detail + )); + } + return Ok(()); + } + if response.status() != StatusCode::TOO_MANY_REQUESTS { + return Err(remote_cache_http_error(response, "upload").await); + } + let limit = read_limit_reached(response).await; + // A wait this short is the server's room for uploads at once, not a + // spent limit: the upload goes again rather than being dropped, + // unless uploads were paused meanwhile. + if !retried && limit.wait <= MAX_RETRY_WAIT { + retried = true; + tokio::time::sleep(limit.wait + retry_jitter(key, limit.wait)).await; + if self.limits.uploads_refused() { + return Ok(()); + } + if self.limits.uploads_paused() { + self.skipping_while_busy(LimitedRequest::Upload); + return Ok(()); + } + continue; + } + self.limit_reached(LimitedRequest::Upload, limit); return Ok(()); } + } - Err(remote_cache_http_error(response, "upload").await) + /// Pauses the refused kind of request for as long as the cache server + /// asks, and warns the first time the run reaches that limit. + fn limit_reached(&self, request: LimitedRequest, limit: LimitReached) { + // A pause of a few seconds is the server's room for requests at once, + // not a spent limit: it gets its own notice, leaving the warning for a + // longer limit later. + if limit.wait <= MAX_RETRY_WAIT { + self.limits.pause(request, limit.wait); + self.skipping_while_busy(request); + return; + } + self.limits.spend(request, limit.wait); + if !self.limits.first_warning(request) { + return; + } + let skipped = match request { + LimitedRequest::Lookup if !self.read_only => "lookups and uploads", + LimitedRequest::Lookup => "lookups", + LimitedRequest::Upload => "uploads", + }; + install_ui::warn_untrusted(&format!( + "{}; skipping remote cache {skipped} for {}", + limit.message, + format_wait(limit.wait), + )); + } + + /// Notes, once a run for each kind, that requests are being skipped + /// while the cache server is busy for a moment. A request paused by a + /// spent limit has that limit's warning instead. + fn skipping_while_busy(&self, request: LimitedRequest) { + if self.limits.paused_by_a_limit(request) || !self.limits.first_busy_notice(request) { + return; + } + install_ui::skipped(busy_notice(request)); } fn artifact_url(&self, key: &str) -> Result { @@ -680,17 +999,25 @@ fn resolve_client( ); } + let team = std::env::var("LPM_REMOTE_CACHE_TEAM") + .ok() + .filter(|team| !team.trim().is_empty()) + .or_else(|| config.and_then(|cfg| cfg.team.clone())); + let limits = RequestLimits::for_cache(&base_url, team.as_deref()); Ok(Some(RemoteCacheClient { base_url, auth, - team: std::env::var("LPM_REMOTE_CACHE_TEAM") - .ok() - .filter(|team| !team.trim().is_empty()) - .or_else(|| config.and_then(|cfg| cfg.team.clone())), + team, signature_key, read_only: env_flag("LPM_REMOTE_CACHE_READ_ONLY").unwrap_or(false) || config.is_some_and(|cfg| cfg.read_only), env_policy: config.map(|cfg| cfg.env.clone()).unwrap_or_default(), + max_artifact_bytes: if cache_matches_registry { + REGISTRY_MAX_ARTIFACT_BYTES + } else { + MAX_REMOTE_ARTIFACT_BYTES + }, + limits, })) } @@ -852,7 +1179,90 @@ fn status_body_too_large() -> String { ) } +/// Reads a 429's message and wait: `Retry-After`, in seconds or as a date, +/// else the body's `retryAfter` seconds, else a minute; at least a second and +/// at most a day. +async fn read_limit_reached(response: reqwest::Response) -> LimitReached { + let header_wait = response + .headers() + .get(reqwest::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| parse_retry_after(value, SystemTime::now())); + let body = read_remote_status_body(response).await.ok(); + let message = body + .as_ref() + .and_then(|body| body.get("error")) + .and_then(serde_json::Value::as_str) + .unwrap_or("remote cache limit reached") + .to_owned(); + let wait = header_wait + .or_else(|| { + body.as_ref() + .and_then(|body| body.get("retryAfter")) + .and_then(serde_json::Value::as_f64) + .and_then(|seconds| Duration::try_from_secs_f64(seconds).ok()) + }) + .unwrap_or(DEFAULT_LIMIT_PAUSE) + .clamp(Duration::from_secs(1), MAX_LIMIT_PAUSE); + LimitReached { message, wait } +} + +fn parse_retry_after(value: &str, now: SystemTime) -> Option { + let value = value.trim(); + if let Ok(seconds) = value.parse::() { + return Some(Duration::from_secs(seconds)); + } + let date = httpdate::parse_http_date(value).ok()?; + Some(date.duration_since(now).unwrap_or(Duration::ZERO)) +} + +/// A wait in the largest whole unit that fits, rounded up. +fn format_wait(wait: Duration) -> String { + let seconds = wait.as_secs() + u64::from(wait.subsec_nanos() > 0); + let (count, unit) = if seconds < 60 { + (seconds, "second") + } else if seconds < 60 * 60 { + (seconds.div_ceil(60), "minute") + } else { + (seconds.div_ceil(60 * 60), "hour") + }; + if count == 1 { + format!("1 {unit}") + } else { + format!("{count} {unit}s") + } +} + async fn remote_cache_http_error(response: reqwest::Response, operation: &str) -> String { + let status = response.status(); + if status == StatusCode::TOO_MANY_REQUESTS { + let limit = read_limit_reached(response).await; + return format!( + "{}; try again in {}", + limit.message, + format_wait(limit.wait) + ); + } + let refusal = read_refusal(response, operation).await; + if operation == "status" { + refusal.detail + } else { + format!("{}; continuing without it", refusal.detail) + } +} + +/// The code the cache server gives an upload it couldn't make room for: +/// a smaller or later upload may still fit. +const QUOTA_EXCEEDED_CODE: &str = "REMOTE_CACHE_QUOTA_EXCEEDED"; + +/// A refused remote cache request: what to tell the user, and the server's +/// code for it, if any. +struct Refusal { + detail: String, + code: Option, +} + +async fn read_refusal(response: reqwest::Response, operation: &str) -> Refusal { let status = response.status(); let body = if matches!(status, StatusCode::FORBIDDEN | StatusCode::PAYMENT_REQUIRED) { read_remote_status_body(response).await.ok() @@ -883,13 +1293,88 @@ async fn remote_cache_http_error(response: reqwest::Response, operation: &str) - } else { format!("remote cache {operation} returned HTTP {}", status.as_u16()) }; - if operation == "status" { - detail - } else { - format!("{detail}; continuing without it") + Refusal { + detail, + code: code.map(ToOwned::to_owned), + } +} + +/// The value of an artifact header, if it's text. +fn artifact_header(response: &reqwest::Response, name: &str) -> Option { + response + .headers() + .get(name) + .and_then(|value| value.to_str().ok()) + .map(ToOwned::to_owned) +} + +const MISSING_CONTENT_HASH: &str = + "remote cache artifact is missing a content hash; treating it as a miss"; +const MISSING_SIGNATURE_TAG: &str = + "remote cache artifact is missing a signature tag; treating it as a miss"; + +/// Fetches an artifact a lookup was redirected to, without the token: the +/// location is a short-lived URL on the cache's download host, which the +/// redirect's own headers vouch for, and a redirect from HTTPS may not lead +/// to plain HTTP. Returns None for an artifact the host no longer serves, a +/// miss. The URL carries the download's credential, so its errors leave it +/// out. +async fn fetch_redirected_artifact( + lookup_url: &reqwest::Url, + redirect: &reqwest::Response, +) -> Result, String> { + let target = redirect_target( + lookup_url, + redirect + .headers() + .get(LOCATION) + .and_then(|value| value.to_str().ok()), + )?; + let response = http_client()?.get(target).send().await.map_err(|error| { + format!( + "remote cache download failed: {}", + lpm_http::display_error(&error.without_url()) + ) + })?; + // The download host's refusals aren't the cache server's: a 403 there + // is the URL's, which may have expired, not the token's. + match response.status() { + StatusCode::OK => Ok(Some(response)), + StatusCode::NOT_FOUND => Ok(None), + status => Err(format!( + "remote cache download returned HTTP {}; continuing without it", + status.as_u16() + )), } } +/// Where a lookup's redirect leads: its location, resolved against the +/// lookup's URL, which may not leave HTTPS. +fn redirect_target( + lookup_url: &reqwest::Url, + location: Option<&str>, +) -> Result { + let location = + location.ok_or("remote cache redirect has no location; treating it as a miss")?; + let target = lookup_url + .join(location) + .map_err(|e| format!("remote cache redirect has an invalid location: {e}"))?; + if lookup_url.scheme() == "https" && target.scheme() != "https" { + return Err("remote cache redirect leaves HTTPS; treating it as a miss".into()); + } + Ok(target) +} + +/// The client for lookups, which leaves redirects to +/// `fetch_redirected_artifact`, so the token never follows one. +fn lookup_client() -> Result { + lpm_http::client_builder() + .redirect(reqwest::redirect::Policy::none()) + .timeout(REMOTE_CACHE_TIMEOUT) + .build() + .map_err(|e| format!("failed to initialize remote cache HTTP client: {e}")) +} + fn http_client() -> Result { lpm_http::client_builder() .timeout(REMOTE_CACHE_TIMEOUT) @@ -990,27 +1475,56 @@ fn env_flag(name: &str) -> Option { } } +fn busy_notice(request: LimitedRequest) -> &'static str { + match request { + LimitedRequest::Lookup => "remote cache is busy; skipping some lookups for a moment", + LimitedRequest::Upload => "remote cache is busy; skipping some uploads for a moment", + } +} + +/// Seeded afresh each run, so CI jobs refused together on the same key +/// don't all go again together either. +static RETRY_JITTER_SEED: LazyLock = LazyLock::new(RandomState::new); + +/// Up to half the wait again, from the artifact's key, so requests refused +/// together don't all go again together. +fn retry_jitter(key: &str, wait: Duration) -> Duration { + let spread = RETRY_JITTER_SEED.hash_one(key); + wait.mul_f64((spread % 1000) as f64 / 2000.0) +} + +/// An artifact size cap, a whole number of MiB, as the Registry states it. +fn format_cap(bytes: u64) -> String { + format!("{} MiB", bytes / (1024 * 1024)) +} + +fn oversized_download(max_artifact_bytes: u64) -> String { + format!( + "remote cache artifact exceeds the {} limit; treating it as a miss", + format_cap(max_artifact_bytes), + ) +} + async fn copy_response_to_temp( response: &mut reqwest::Response, temp: &mut tempfile::NamedTempFile, signature_key: Option<&str>, + max_artifact_bytes: u64, ) -> Result { let mut total = 0u64; let mut sha = Sha256::new(); let mut mac = signature_key.map(new_hmac).transpose()?; - while let Some(buffer) = response - .chunk() - .await - .map_err(|e| format!("failed to read remote cache artifact: {e}"))? - { + while let Some(buffer) = response.chunk().await.map_err(|error| { + format!( + "failed to read remote cache artifact: {}", + error.without_url() + ) + })? { let read = buffer.len(); total = total.saturating_add(read as u64); - if total > MAX_REMOTE_ARTIFACT_BYTES { - return Err(format!( - "remote cache artifact exceeds the {} limit; treating it as a miss", - format_bytes(MAX_REMOTE_ARTIFACT_BYTES), - )); + if total > max_artifact_bytes { + return Err(oversized_download(max_artifact_bytes)); } sha.update(&buffer[..read]); @@ -1490,6 +2004,961 @@ mod tests { assert!(reason.contains("CI_DEPLOY_PASSWORD")); } + #[test] + fn retry_after_reads_seconds_and_dates() { + let now = httpdate::parse_http_date("Sun, 11 Oct 2026 12:00:00 GMT").unwrap(); + + assert_eq!( + parse_retry_after(" 42 ", now), + Some(Duration::from_secs(42)) + ); + assert_eq!( + parse_retry_after("Sun, 11 Oct 2026 12:01:30 GMT", now), + Some(Duration::from_secs(90)) + ); + assert_eq!( + parse_retry_after("Sun, 11 Oct 2026 11:59:00 GMT", now), + Some(Duration::ZERO) + ); + assert_eq!(parse_retry_after("soon", now), None); + assert_eq!(parse_retry_after("-5", now), None); + } + + #[test] + fn waits_read_in_their_largest_unit_rounded_up() { + for (seconds, expected) in [ + (0.5, "1 second"), + (1.0, "1 second"), + (42.0, "42 seconds"), + (60.0, "1 minute"), + (61.0, "2 minutes"), + (3_600.0, "1 hour"), + (86_339.0, "24 hours"), + ] { + assert_eq!(format_wait(Duration::from_secs_f64(seconds)), expected); + } + } + + async fn limit_from(response: wiremock::ResponseTemplate) -> LimitReached { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::any()) + .respond_with(response) + .mount(&server) + .await; + let response = reqwest::get(server.uri()).await.unwrap(); + read_limit_reached(response).await + } + + #[tokio::test] + async fn a_limit_waits_as_long_as_the_server_asks_within_a_second_and_a_day() { + let limited = || { + wiremock::ResponseTemplate::new(429).set_body_json(serde_json::json!({ + "error": "Remote cache limit reached: 300 uploads a minute", + "retryAfter": 30, + })) + }; + + let from_header = limit_from(limited().insert_header("retry-after", "42")).await; + assert_eq!( + from_header.message, + "Remote cache limit reached: 300 uploads a minute" + ); + assert_eq!(from_header.wait, Duration::from_secs(42)); + assert_eq!( + limit_from(limited()).await.wait, + Duration::from_secs(30), + "without the header, the body's retryAfter" + ); + assert_eq!( + limit_from(limited().insert_header("retry-after", "0")) + .await + .wait, + Duration::from_secs(1) + ); + assert_eq!( + limit_from(limited().insert_header("retry-after", "604800")) + .await + .wait, + MAX_LIMIT_PAUSE + ); + + let bare = limit_from(wiremock::ResponseTemplate::new(429)).await; + assert_eq!(bare.message, "remote cache limit reached"); + assert_eq!(bare.wait, DEFAULT_LIMIT_PAUSE); + } + + #[test] + fn a_lookup_limit_pauses_uploads_too_but_an_upload_limit_pauses_only_uploads() { + let lookups = RequestLimits::default(); + lookups.pause(LimitedRequest::Lookup, Duration::from_secs(60)); + assert!(lookups.lookups_paused() && lookups.uploads_paused()); + + let uploads = RequestLimits::default(); + uploads.pause(LimitedRequest::Upload, Duration::from_secs(60)); + assert!(!uploads.lookups_paused() && uploads.uploads_paused()); + } + + #[test] + fn a_pause_ends_when_its_wait_does_and_a_shorter_one_never_cuts_it() { + let limits = RequestLimits::default(); + limits.pause(LimitedRequest::Lookup, Duration::ZERO); + assert!(!limits.lookups_paused()); + + limits.pause(LimitedRequest::Lookup, Duration::from_secs(60)); + limits.pause(LimitedRequest::Lookup, Duration::ZERO); + assert!(limits.lookups_paused()); + } + + #[test] + fn each_kind_of_limit_warns_once() { + let limits = RequestLimits::default(); + + assert!(limits.first_warning(LimitedRequest::Lookup)); + assert!(!limits.first_warning(LimitedRequest::Lookup)); + assert!(limits.first_warning(LimitedRequest::Upload)); + assert!(!limits.first_warning(LimitedRequest::Upload)); + } + + #[test] + fn clients_share_limits_per_cache_server_and_team() { + let url = "https://limits-test.example/v8"; + let shared = RequestLimits::for_cache(url, Some("team-a")); + + assert!(Arc::ptr_eq( + &shared, + &RequestLimits::for_cache(url, Some("team-a")) + )); + assert!(!Arc::ptr_eq( + &shared, + &RequestLimits::for_cache(url, Some("team-b")) + )); + assert!(!Arc::ptr_eq(&shared, &RequestLimits::for_cache(url, None))); + assert!(!Arc::ptr_eq( + &shared, + &RequestLimits::for_cache("https://other.example/v8", Some("team-a")) + )); + } + + const MIB: u64 = 1024 * 1024; + + fn test_client(base_url: &str, max_artifact_bytes: u64) -> RemoteCacheClient { + RemoteCacheClient { + base_url: base_url.to_owned(), + auth: RemoteCacheAuth::Static("remote-token".into()), + team: None, + signature_key: None, + read_only: false, + env_policy: lpm_runner::lpm_json::RemoteCacheEnvConfig::default(), + max_artifact_bytes, + limits: Arc::new(RequestLimits::default()), + } + } + + #[test] + fn uploads_only_artifacts_within_the_cache_servers_cap() { + let client = test_client("http://127.0.0.1:9/v8", MIB); + + assert_eq!(client.check_upload_size(MIB), Ok(())); + assert_eq!( + client.check_upload_size(MIB + 1), + Err("remote cache artifact exceeds the 1 MiB upload limit; skipping upload".into()) + ); + } + + /// A cache server that streams `bytes` of every artifact in chunks, with + /// no Content-Length, so only the streaming check can stop a large one. + fn chunked_artifact_server(bytes: u64) -> String { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + std::thread::spawn(move || { + for mut stream in listener.incoming().flatten() { + let mut head = Vec::new(); + let mut byte = [0u8; 1]; + while !head.ends_with(b"\r\n\r\n") && matches!(stream.read(&mut byte), Ok(1)) { + head.push(byte[0]); + } + let chunk = vec![0u8; 64 * 1024]; + let mut sent = 0; + let mut send = |data: &[u8]| stream.write_all(data).is_ok(); + if !send( + b"HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\nconnection: close\r\n\r\n", + ) { + continue; + } + while sent < bytes + && send(format!("{:x}\r\n", chunk.len()).as_bytes()) + && send(&chunk) + && send(b"\r\n") + { + sent += chunk.len() as u64; + } + send(b"0\r\n\r\n"); + } + }); + url + } + + #[tokio::test] + async fn treats_a_streamed_artifact_past_the_cache_servers_cap_as_a_miss() { + let url = chunked_artifact_server(2 * MIB); + let client = test_client(&format!("{url}/v8"), MIB); + let project = tempfile::tempdir().unwrap(); + + let restored = client + .restore_async(&"a".repeat(64), project.path(), &[], || Ok(true)) + .await; + + match restored { + Err(error) => assert_eq!( + error, + "remote cache artifact exceeds the 1 MiB limit; treating it as a miss" + ), + Ok(_) => panic!("an artifact past the cap was restored"), + } + } + + async fn upload_to(server: &wiremock::MockServer) -> (RemoteCacheClient, Result<(), String>) { + let client = test_client(&format!("{}/v8", server.uri()), MIB); + let mut artifact = tempfile::tempfile().unwrap(); + artifact.write_all(b"artifact").unwrap(); + let uploaded = client + .upload_artifact(&"a".repeat(64), &artifact, 8, 10) + .await; + (client, uploaded) + } + + fn upload_refused(retry_after: &str) -> wiremock::ResponseTemplate { + wiremock::ResponseTemplate::new(429) + .insert_header("retry-after", retry_after) + .set_body_json(serde_json::json!({ + "error": "Remote cache limit reached: 8 uploads at once", + })) + } + + #[tokio::test] + async fn an_upload_the_token_may_not_make_stops_uploads_for_the_run() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with( + wiremock::ResponseTemplate::new(403).set_body_json(serde_json::json!({ + "error": "Token does not have remote cache write permissions", + })), + ) + .expect(1) + .mount(&server) + .await; + + let (client, uploaded) = upload_to(&server).await; + + assert_eq!(uploaded, Ok(())); + assert!(client.limits.uploads_refused()); + assert!(!client.limits.lookups_refused()); + assert!(!client.limits.uploads_paused()); + let client = client.clone(); + tokio::task::spawn_blocking(move || store_with(&client)) + .await + .unwrap(); + server.verify().await; + } + + #[tokio::test] + async fn a_lookup_the_token_may_not_make_stops_the_cache_for_the_run() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .respond_with( + wiremock::ResponseTemplate::new(403).set_body_json(serde_json::json!({ + "error": "Token does not have remote cache read permissions", + })), + ) + .expect(1) + .mount(&server) + .await; + let client = test_client(&format!("{}/v8", server.uri()), MIB); + let project = tempfile::tempdir().unwrap(); + + let restored = client + .restore_async(&"a".repeat(64), project.path(), &[], || Ok(true)) + .await; + + assert!(matches!(restored, Ok(None))); + assert!(client.limits.lookups_refused() && client.limits.uploads_refused()); + let client = client.clone(); + tokio::task::spawn_blocking(move || { + let project = tempfile::tempdir().unwrap(); + let restored = try_restore(&client, &"b".repeat(64), project.path(), &[], || Ok(true)); + assert!(restored.is_none()); + store_with(&client); + }) + .await + .unwrap(); + server.verify().await; + } + + fn billing_lapsed() -> wiremock::ResponseTemplate { + wiremock::ResponseTemplate::new(402).set_body_json(serde_json::json!({ + "error": "Organization billing requires attention.", + "code": "ORG_BILLING_LAPSED", + "status": "paused", + })) + } + + #[tokio::test] + async fn an_upload_the_token_may_not_make_leaves_lookups_on() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(wiremock::ResponseTemplate::new(403).set_body_json( + serde_json::json!({ "error": "Token does not have remote cache write permissions" }), + )) + .mount(&server) + .await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .respond_with(wiremock::ResponseTemplate::new(404)) + .expect(1) + .mount(&server) + .await; + let (client, _) = upload_to(&server).await; + let project = tempfile::tempdir().unwrap(); + + let restored = client + .restore_async(&"b".repeat(64), project.path(), &[], || Ok(true)) + .await; + + assert!(matches!(restored, Ok(None))); + server.verify().await; + } + + #[test] + fn a_redirect_resolves_against_the_lookup_and_may_not_leave_https() { + let lookup = reqwest::Url::parse("https://lpm.dev/v8/artifacts/abc?slug=acme").unwrap(); + + assert_eq!( + redirect_target(&lookup, Some("/storage/abc")) + .unwrap() + .as_str(), + "https://lpm.dev/storage/abc" + ); + assert_eq!( + redirect_target(&lookup, Some("https://storage.example/abc?sig=1")) + .unwrap() + .as_str(), + "https://storage.example/abc?sig=1" + ); + assert!( + redirect_target(&lookup, Some("http://storage.example/abc")) + .unwrap_err() + .contains("leaves HTTPS") + ); + assert!( + redirect_target(&lookup, None) + .unwrap_err() + .contains("no location") + ); + let plain = reqwest::Url::parse("http://127.0.0.1:3000/v8/artifacts/abc").unwrap(); + assert!(redirect_target(&plain, Some("http://127.0.0.1:3000/storage/abc")).is_ok()); + } + + #[tokio::test] + async fn a_lapsed_bill_on_a_lookup_stops_the_cache_for_the_run() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .respond_with(billing_lapsed()) + .expect(1) + .mount(&server) + .await; + let client = test_client(&format!("{}/v8", server.uri()), MIB); + let project = tempfile::tempdir().unwrap(); + + let restored = client + .restore_async(&"a".repeat(64), project.path(), &[], || Ok(true)) + .await; + + assert!(matches!(restored, Ok(None))); + assert!(client.limits.lookups_refused() && client.limits.uploads_refused()); + let client = client.clone(); + tokio::task::spawn_blocking(move || { + let project = tempfile::tempdir().unwrap(); + assert!( + try_restore(&client, &"b".repeat(64), project.path(), &[], || Ok(true)).is_none() + ); + store_with(&client); + }) + .await + .unwrap(); + server.verify().await; + } + + #[tokio::test] + async fn a_lapsed_bill_on_an_upload_stops_the_cache_for_the_run() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(billing_lapsed()) + .expect(1) + .mount(&server) + .await; + + let (client, uploaded) = upload_to(&server).await; + + assert_eq!(uploaded, Ok(())); + assert!(client.limits.lookups_refused() && client.limits.uploads_refused()); + server.verify().await; + } + + #[tokio::test] + async fn an_upload_the_cache_has_no_room_for_leaves_uploads_on() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with( + wiremock::ResponseTemplate::new(403).set_body_json(serde_json::json!({ + "error": "Remote cache quota exceeded", + "status": "over_limit", + "code": "REMOTE_CACHE_QUOTA_EXCEEDED", + })), + ) + .expect(1) + .mount(&server) + .await; + + let (client, uploaded) = upload_to(&server).await; + + assert_eq!( + uploaded, + Err("remote cache quota was exceeded; free cache space or enable overage; continuing without it".to_string()) + ); + assert!(!client.limits.uploads_refused()); + server.verify().await; + } + + #[tokio::test] + async fn an_upload_of_an_account_over_its_limit_stops_uploads_for_the_run() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(wiremock::ResponseTemplate::new(403).set_body_json( + serde_json::json!({ "error": "Remote cache is not enabled", "status": "over_limit" }), + )) + .mount(&server) + .await; + + let (client, uploaded) = upload_to(&server).await; + + assert_eq!(uploaded, Ok(())); + assert!(client.limits.uploads_refused()); + } + + #[tokio::test] + async fn an_upload_waiting_to_go_again_is_dropped_once_the_run_stops_uploads() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(upload_refused("1")) + .expect(1) + .mount(&server) + .await; + let client = test_client(&format!("{}/v8", server.uri()), MIB); + let limits = client.limits.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(200)).await; + limits.refuse_uploads(); + }); + let mut artifact = tempfile::tempfile().unwrap(); + artifact.write_all(b"artifact").unwrap(); + + let uploaded = client + .upload_artifact(&"a".repeat(64), &artifact, 8, 10) + .await; + + assert_eq!(uploaded, Ok(())); + server.verify().await; + } + + #[tokio::test] + async fn a_lookup_waiting_to_go_again_is_dropped_once_the_run_stops_lookups() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .respond_with(upload_refused("1")) + .expect(1) + .mount(&server) + .await; + let client = test_client(&format!("{}/v8", server.uri()), MIB); + let limits = client.limits.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(200)).await; + limits.refuse_lookups(); + }); + let project = tempfile::tempdir().unwrap(); + + let restored = client + .restore_async(&"a".repeat(64), project.path(), &[], || Ok(true)) + .await; + + assert!(matches!(restored, Ok(None))); + server.verify().await; + } + + #[test] + fn says_once_that_the_run_stops_each_kind_of_request() { + let limits = RequestLimits::default(); + + assert!(limits.refuse_uploads()); + assert!(!limits.refuse_uploads()); + assert!(limits.refuse_lookups()); + assert!(!limits.refuse_lookups()); + assert!(!limits.refuse_uploads()); + } + + #[tokio::test] + async fn an_upload_refused_for_another_reason_leaves_uploads_on() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(wiremock::ResponseTemplate::new(500)) + .mount(&server) + .await; + + let (client, uploaded) = upload_to(&server).await; + + assert!(uploaded.is_err()); + assert!(!client.limits.uploads_refused()); + } + + #[tokio::test] + async fn an_upload_refused_for_a_moment_goes_once_more_and_pauses_nothing() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(upload_refused("1")) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with( + wiremock::ResponseTemplate::new(200) + .set_body_json(serde_json::json!({ "urls": [] })), + ) + .expect(1) + .mount(&server) + .await; + + let (client, uploaded) = upload_to(&server).await; + + assert_eq!(uploaded, Ok(())); + assert!(!client.limits.uploads_paused()); + server.verify().await; + } + + #[tokio::test] + async fn an_upload_another_is_storing_is_left_to_it_quietly() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(wiremock::ResponseTemplate::new(409).set_body_json( + serde_json::json!({ "error": "Remote cache artifact upload already in progress" }), + )) + .expect(1) + .mount(&server) + .await; + + let (client, uploaded) = upload_to(&server).await; + + assert_eq!(uploaded, Ok(())); + assert!(!client.limits.uploads_paused()); + server.verify().await; + } + + #[tokio::test] + async fn uploads_pause_once_a_retry_is_refused_too_or_the_wait_is_a_spent_limit() { + for (retry_after, attempts) in [("1", 2), ("6", 1), ("60", 1)] { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(upload_refused(retry_after)) + .expect(attempts) + .mount(&server) + .await; + + let (client, uploaded) = upload_to(&server).await; + + assert_eq!(uploaded, Ok(())); + assert!(client.limits.uploads_paused(), "Retry-After {retry_after}"); + server.verify().await; + } + } + + #[tokio::test] + async fn an_upload_refused_for_five_seconds_still_goes_once_more() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(upload_refused("5")) + .expect(2) + .mount(&server) + .await; + + let (client, uploaded) = upload_to(&server).await; + + assert_eq!(uploaded, Ok(())); + assert!(client.limits.uploads_paused()); + server.verify().await; + } + + #[tokio::test] + async fn an_upload_doesnt_go_again_once_uploads_or_lookups_were_paused_while_it_waited() { + for paused in [LimitedRequest::Upload, LimitedRequest::Lookup] { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(upload_refused("1")) + .expect(1) + .mount(&server) + .await; + let client = test_client(&format!("{}/v8", server.uri()), MIB); + let mut artifact = tempfile::tempfile().unwrap(); + artifact.write_all(b"artifact").unwrap(); + let limits = client.limits.clone(); + let pause = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(300)).await; + limits.spend(paused, Duration::from_secs(60)); + }); + + let uploaded = client + .upload_artifact(&"a".repeat(64), &artifact, 8, 10) + .await; + + pause.await.unwrap(); + assert_eq!(uploaded, Ok(())); + server.verify().await; + // A minute's pause is a spent limit, whose warning covers the + // dropped upload, so the busy notice stays unspent. + assert!(client.limits.first_busy_notice(LimitedRequest::Upload)); + } + } + + fn store_with(client: &RemoteCacheClient) { + let project = tempfile::tempdir().unwrap(); + try_store( + client, + "abc123", + project.path(), + "build", + &[], + "", + "", + 1, + &HashMap::new(), + &HashMap::new(), + ); + } + + #[test] + fn an_upload_skipped_while_a_lookup_waits_on_a_busy_server_gets_the_busy_notice() { + let client = test_client("http://127.0.0.1:9/v8", MIB); + client + .limits + .pause(LimitedRequest::Lookup, Duration::from_secs(3)); + + store_with(&client); + + assert!(!client.limits.first_busy_notice(LimitedRequest::Upload)); + } + + #[test] + fn an_upload_skipped_for_a_spent_lookup_limit_has_that_limit_s_warning_instead() { + let client = test_client("http://127.0.0.1:9/v8", MIB); + client + .limits + .spend(LimitedRequest::Lookup, Duration::from_secs(60)); + + store_with(&client); + + assert!(client.limits.first_busy_notice(LimitedRequest::Upload)); + } + + #[test] + fn a_read_only_client_skips_uploads_without_a_notice() { + let mut client = test_client("http://127.0.0.1:9/v8", MIB); + client.read_only = true; + client + .limits + .pause(LimitedRequest::Upload, Duration::from_secs(3)); + + store_with(&client); + + assert!(client.limits.first_busy_notice(LimitedRequest::Upload)); + } + + #[test] + fn a_lookup_skipped_while_the_server_is_busy_gets_the_busy_notice() { + let client = test_client("http://127.0.0.1:9/v8", MIB); + client + .limits + .pause(LimitedRequest::Lookup, Duration::from_secs(3)); + let project = tempfile::tempdir().unwrap(); + + let hit = try_restore(&client, "abc123", project.path(), &[], || Ok(true)); + + assert!(hit.is_none()); + assert!(!client.limits.first_busy_notice(LimitedRequest::Lookup)); + } + + #[test] + fn a_key_s_retry_jitter_holds_for_the_whole_run() { + let wait = Duration::from_secs(2); + for index in 0..32 { + let key = format!("{index:064x}"); + let jitter = retry_jitter(&key, wait); + assert_eq!(retry_jitter(&key, wait), jitter); + assert!(jitter < wait / 2, "{jitter:?}"); + } + } + + #[test] + fn the_busy_notice_names_what_it_skips_in_lpm_s_own_words() { + assert_eq!( + busy_notice(LimitedRequest::Lookup), + "remote cache is busy; skipping some lookups for a moment" + ); + assert_eq!( + busy_notice(LimitedRequest::Upload), + "remote cache is busy; skipping some uploads for a moment" + ); + } + + #[test] + fn a_short_refusal_during_a_spent_limit_leaves_the_busy_notice_unspent() { + let client = test_client("http://127.0.0.1:9/v8", MIB); + client + .limits + .spend(LimitedRequest::Lookup, Duration::from_secs(60)); + + client.limit_reached( + LimitedRequest::Upload, + LimitReached { + message: "Remote cache limit reached: 8 uploads at once".into(), + wait: Duration::from_secs(1), + }, + ); + + assert!(client.limits.first_busy_notice(LimitedRequest::Upload)); + } + + fn lookup_refused(retry_after: &str) -> wiremock::ResponseTemplate { + wiremock::ResponseTemplate::new(429) + .insert_header("retry-after", retry_after) + .set_body_json(serde_json::json!({ + "error": "Remote cache limit reached: 32 uploads and downloads at once", + })) + } + + async fn look_up_at(client: &RemoteCacheClient) { + let project = tempfile::tempdir().unwrap(); + let restored = client + .restore_async(&"a".repeat(64), project.path(), &[], || Ok(true)) + .await; + assert!(matches!(restored, Ok(None))); + } + + #[tokio::test] + async fn a_lookup_refused_for_a_moment_goes_once_more_and_pauses_nothing() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .respond_with(lookup_refused("1")) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .respond_with(wiremock::ResponseTemplate::new(404)) + .expect(1) + .mount(&server) + .await; + let client = test_client(&format!("{}/v8", server.uri()), MIB); + let limits = client.limits.clone(); + let during = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(500)).await; + limits.lookups_paused() || limits.uploads_paused() + }); + + look_up_at(&client).await; + + assert!(!during.await.unwrap(), "nothing pauses while it waits"); + assert!(!client.limits.lookups_paused()); + server.verify().await; + } + + #[tokio::test] + async fn a_lookup_pauses_lookups_once_its_retry_is_refused_too() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .respond_with(lookup_refused("1")) + .expect(2) + .mount(&server) + .await; + let client = test_client(&format!("{}/v8", server.uri()), MIB); + + look_up_at(&client).await; + + assert!(client.limits.lookups_paused()); + assert!(!client.limits.first_busy_notice(LimitedRequest::Lookup)); + server.verify().await; + } + + #[tokio::test] + async fn a_lookup_doesnt_go_again_once_lookups_were_paused_while_it_waited() { + // A spent limit's warning covers the skipped lookup; a busy server's + // pause gets the busy notice. + for (spent, noted) in [(true, false), (false, true)] { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .respond_with(lookup_refused("1")) + .expect(1) + .mount(&server) + .await; + let client = test_client(&format!("{}/v8", server.uri()), MIB); + let limits = client.limits.clone(); + let pause = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(300)).await; + if spent { + limits.spend(LimitedRequest::Lookup, Duration::from_secs(60)); + } else { + limits.pause(LimitedRequest::Lookup, Duration::from_secs(3)); + } + }); + + look_up_at(&client).await; + + pause.await.unwrap(); + server.verify().await; + assert_eq!( + !client.limits.first_busy_notice(LimitedRequest::Lookup), + noted, + "spent limit: {spent}" + ); + } + } + + #[tokio::test] + async fn a_lookup_refused_for_five_seconds_still_goes_once_more() { + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .respond_with(lookup_refused("5")) + .expect(2) + .mount(&server) + .await; + let client = test_client(&format!("{}/v8", server.uri()), MIB); + + look_up_at(&client).await; + + assert!(client.limits.lookups_paused()); + assert!(!client.limits.paused_by_a_limit(LimitedRequest::Lookup)); + server.verify().await; + } + + #[tokio::test] + async fn a_lookup_goes_again_after_the_wait_and_its_share_of_the_spread() { + let wait = Duration::from_secs(1); + let key = (0u32..) + .map(|index| format!("{index:064x}")) + .find(|key| retry_jitter(key, wait) >= Duration::from_millis(300)) + .unwrap(); + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .respond_with(lookup_refused("1")) + .up_to_n_times(1) + .mount(&server) + .await; + wiremock::Mock::given(wiremock::matchers::method("GET")) + .respond_with(wiremock::ResponseTemplate::new(404)) + .mount(&server) + .await; + let client = test_client(&format!("{}/v8", server.uri()), MIB); + let project = tempfile::tempdir().unwrap(); + + let started = Instant::now(); + let restored = client + .restore_async(&key, project.path(), &[], || Ok(true)) + .await; + + assert!(matches!(restored, Ok(None))); + assert!(started.elapsed() >= wait + retry_jitter(&key, wait)); + } + + #[test] + fn a_spent_limit_is_told_from_a_busy_server_however_little_of_it_is_left() { + let limits = RequestLimits::default(); + limits.spend(LimitedRequest::Lookup, Duration::from_millis(5_500)); + std::thread::sleep(Duration::from_millis(600)); + assert!(limits.paused_by_a_limit(LimitedRequest::Lookup)); + assert!(limits.paused_by_a_limit(LimitedRequest::Upload)); + + let busy = RequestLimits::default(); + busy.pause(LimitedRequest::Lookup, Duration::from_secs(60)); + assert!(busy.lookups_paused()); + assert!(!busy.paused_by_a_limit(LimitedRequest::Lookup)); + + let uploads = RequestLimits::default(); + uploads.spend(LimitedRequest::Upload, Duration::from_secs(60)); + assert!(!uploads.paused_by_a_limit(LimitedRequest::Lookup)); + } + + #[tokio::test] + async fn an_upload_goes_again_after_the_wait_and_its_share_of_the_spread() { + let wait = Duration::from_secs(1); + let key = (0u32..) + .map(|index| format!("{index:064x}")) + .find(|key| retry_jitter(key, wait) >= Duration::from_millis(300)) + .unwrap(); + let server = wiremock::MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with(upload_refused("1")) + .up_to_n_times(1) + .mount(&server) + .await; + wiremock::Mock::given(wiremock::matchers::method("PUT")) + .respond_with( + wiremock::ResponseTemplate::new(200) + .set_body_json(serde_json::json!({ "urls": [] })), + ) + .mount(&server) + .await; + let client = test_client(&format!("{}/v8", server.uri()), MIB); + let mut artifact = tempfile::tempfile().unwrap(); + artifact.write_all(b"artifact").unwrap(); + + let started = Instant::now(); + let uploaded = client.upload_artifact(&key, &artifact, 8, 10).await; + + assert_eq!(uploaded, Ok(())); + assert!(started.elapsed() >= wait + retry_jitter(&key, wait)); + } + + #[test] + fn retries_spread_over_up_to_half_the_wait_again() { + let wait = Duration::from_secs(1); + let spreads: std::collections::BTreeSet<_> = (0..50) + .map(|index| retry_jitter(&format!("{index:064x}"), wait)) + .collect(); + + assert!(spreads.iter().all(|spread| *spread <= wait / 2)); + assert!(spreads.len() > 10, "{spreads:?}"); + } + + #[test] + fn a_pause_of_up_to_five_seconds_gets_its_own_notice_and_leaves_the_warning_for_a_longer_limit() + { + let limit = |seconds| LimitReached { + message: "Remote cache limit reached: 8 uploads at once".into(), + wait: Duration::from_secs(seconds), + }; + for request in [LimitedRequest::Upload, LimitedRequest::Lookup] { + let client = test_client("http://127.0.0.1:9/v8", MIB); + + client.limit_reached(request, limit(5)); + assert!(client.limits.paused(request)); + assert!(!client.limits.paused_by_a_limit(request)); + assert!(!client.limits.first_busy_notice(request)); + assert!(client.limits.first_warning(request)); + + let client = test_client("http://127.0.0.1:9/v8", MIB); + client.limit_reached(request, limit(6)); + assert!(client.limits.paused_by_a_limit(request)); + assert!(!client.limits.first_warning(request)); + assert!(client.limits.first_busy_notice(request)); + } + } + #[test] fn exclude_pattern_overrides_include_pattern() { let policy = lpm_runner::lpm_json::RemoteCacheEnvConfig { diff --git a/tests/workflows/tests/cache.rs b/tests/workflows/tests/cache.rs index 8885e42e5..06816a402 100644 --- a/tests/workflows/tests/cache.rs +++ b/tests/workflows/tests/cache.rs @@ -177,6 +177,53 @@ async fn cache_status_rejects_an_oversized_remote_response() { ); } +#[tokio::test] +async fn cache_status_reports_a_request_limit_and_when_to_try_again() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/v8/artifacts/status")) + .respond_with( + ResponseTemplate::new(429) + .insert_header("retry-after", "90") + .set_body_json(serde_json::json!({ + "error": "Remote cache limit reached: 3,000 requests a minute", + "retryAfter": 90, + })), + ) + .expect(1) + .mount(&server) + .await; + + let project = TempProject::empty(r#"{"name":"cache-status","version":"1.0.0"}"#); + project.write_file( + "lpm.json", + &format!( + r#"{{ "remoteCache": {{ "enabled": true, "url": "{}/v8" }} }}"#, + server.uri(), + ), + ); + + let output = lpm(&project) + .env("LPM_REMOTE_CACHE_TOKEN", "remote-token") + .env("LPM_REMOTE_CACHE_SIGNATURE_KEY", "signing-key") + .args(["--json", "cache", "status"]) + .output() + .expect("failed to run lpm cache status --json"); + + assert!( + output.status.success(), + "cache status must remain best-effort: {}", + String::from_utf8_lossy(&output.stderr), + ); + let envelope: serde_json::Value = + serde_json::from_slice(&output.stdout).expect("cache status must emit valid JSON"); + assert_eq!( + envelope["remote"]["error"], + "Remote cache limit reached: 3,000 requests a minute; try again in 2 minutes" + ); + server.verify().await; +} + #[test] fn cache_status_fails_when_lpm_json_is_malformed() { let project = TempProject::empty(r#"{"name":"cache-status","version":"1.0.0"}"#); diff --git a/tests/workflows/tests/run.rs b/tests/workflows/tests/run.rs index 8bcdbee72..e5a8d28d0 100644 --- a/tests/workflows/tests/run.rs +++ b/tests/workflows/tests/run.rs @@ -783,6 +783,115 @@ async fn mount_stateful_remote_cache(server: &MockServer, state: RemoteCacheStat .await; } +/// Answers a lookup that asks for a redirect as the LPM.dev Registry does: +/// 302 to the artifact's storage path, with its headers on the redirect. +struct RedirectingGetResponder { + state: RemoteCacheState, + /// Where the redirect leads, in place of this server's storage. + location: Option, +} + +impl Respond for RedirectingGetResponder { + fn respond(&self, request: &Request) -> ResponseTemplate { + let asked = request + .headers + .get("x-lpm-download-redirect") + .is_some_and(|value| value == "1"); + let stored = self + .state + .artifact + .lock() + .expect("remote cache artifact mutex poisoned") + .is_some(); + if !asked || !stored { + return RemoteCacheGetResponder { + state: self.state.clone(), + } + .respond(request); + } + let hash = request.url.path().rsplit('/').next().unwrap_or_default(); + let location = self + .location + .clone() + .unwrap_or_else(|| format!("/storage/{hash}")); + let mut response = ResponseTemplate::new(302).insert_header("Location", location); + if let Some(tag) = self + .state + .tag + .lock() + .expect("remote cache tag mutex poisoned") + .clone() + { + response = response.insert_header("x-artifact-tag", tag); + } + if let Some(sha) = self + .state + .sha + .lock() + .expect("remote cache sha mutex poisoned") + .clone() + { + response = response.insert_header("x-artifact-sha", sha); + } + response + } +} + +/// Serves stored artifacts from storage, refusing a request that carries a +/// token, which a redirect must not pass on. +struct StorageGetResponder { + state: RemoteCacheState, + present: bool, +} + +impl Respond for StorageGetResponder { + fn respond(&self, request: &Request) -> ResponseTemplate { + if request.headers.contains_key("authorization") { + return ResponseTemplate::new(400).set_body_string("token sent to storage"); + } + let artifact = self + .state + .artifact + .lock() + .expect("remote cache artifact mutex poisoned") + .clone(); + match artifact { + Some(artifact) if self.present => ResponseTemplate::new(200) + .insert_header("Content-Length", artifact.len().to_string()) + .set_body_bytes(artifact), + _ => ResponseTemplate::new(404).set_body_string("NoSuchKey"), + } + } +} + +async fn mount_redirecting_remote_cache( + server: &MockServer, + state: RemoteCacheState, + present: bool, +) { + Mock::given(method("GET")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(RedirectingGetResponder { + state: state.clone(), + location: None, + }) + .mount(server) + .await; + Mock::given(method("GET")) + .and(path_regex(r"^/storage/[a-fA-F0-9]+$")) + .respond_with(StorageGetResponder { + state: state.clone(), + present, + }) + .mount(server) + .await; + Mock::given(method("PUT")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(RemoteCachePutResponder { state }) + .mount(server) + .await; +} + fn remote_cache_project(server: &MockServer, extra_remote_config: &str) -> TempProject { let project = TempProject::empty( r#"{ @@ -3099,6 +3208,246 @@ async fn run_remote_cache_miss_uploads_and_later_restores_outputs() { ); } +#[tokio::test] +async fn run_remote_cache_restores_from_the_storage_a_lookup_is_redirected_to_without_the_token() { + let server = MockServer::start().await; + let state = RemoteCacheState::default(); + mount_redirecting_remote_cache(&server, state.clone(), true).await; + let project = remote_cache_project(&server, ""); + + let first = run_build_with_remote_token(&project); + assert!(first.status.success()); + assert!( + state + .artifact + .lock() + .expect("remote cache artifact mutex poisoned") + .is_some(), + "the first run should upload its artifact:\nstderr: {}", + String::from_utf8_lossy(&first.stderr), + ); + remove_local_task_cache(&project); + remove_project_file(&project, "dist"); + remove_project_file(&project, "executed-marker"); + + let second = run_build_with_remote_token(&project); + + assert!( + second.status.success(), + "redirected hit run should succeed:\nstdout: {}\nstderr: {}", + String::from_utf8_lossy(&second.stdout), + String::from_utf8_lossy(&second.stderr), + ); + assert_eq!(project.read_file("dist/value.txt"), "remote-hit"); + assert!( + !project.file_exists("executed-marker"), + "a redirected hit must restore outputs without re-running the script:\nstdout: {}\nstderr: {}", + String::from_utf8_lossy(&second.stdout), + String::from_utf8_lossy(&second.stderr), + ); + let storage_reads = server + .received_requests() + .await + .unwrap() + .into_iter() + .filter(|request| request.url.path().starts_with("/storage/")) + .count(); + assert_eq!(storage_reads, 1); +} + +#[tokio::test] +async fn run_remote_cache_leaves_the_signed_url_out_of_a_failed_download_s_warning() { + const CREDENTIAL: &str = "signed-download-credential-7f3a"; + let server = MockServer::start().await; + let state = RemoteCacheState::default(); + // A port nothing listens on, so the download fails to connect. + let storage = std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap(); + let location = format!("http://{storage}/storage/artifact?verify={CREDENTIAL}"); + Mock::given(method("GET")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(RedirectingGetResponder { + state: state.clone(), + location: Some(location), + }) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(RemoteCachePutResponder { + state: state.clone(), + }) + .mount(&server) + .await; + let project = remote_cache_project(&server, ""); + + assert!(run_build_with_remote_token(&project).status.success()); + remove_local_task_cache(&project); + remove_project_file(&project, "dist"); + remove_project_file(&project, "executed-marker"); + + let second = run_build_with_remote_token(&project); + + let output = format!( + "{}{}", + String::from_utf8_lossy(&second.stdout), + String::from_utf8_lossy(&second.stderr) + ); + assert!(second.status.success(), "{output}"); + assert!(project.file_exists("executed-marker"), "{output}"); + assert!( + output.contains("remote cache download failed"), + "the download should have been tried:\n{output}" + ); + assert!(!output.contains(CREDENTIAL), "{output}"); + assert!(!output.contains("/storage/artifact"), "{output}"); +} + +#[tokio::test] +async fn run_remote_cache_says_a_refused_download_is_the_download_host_s_not_the_token_s() { + let server = MockServer::start().await; + let state = RemoteCacheState::default(); + Mock::given(method("GET")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(RedirectingGetResponder { + state: state.clone(), + location: Some("/expired-download".to_owned()), + }) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path_regex(r"^/expired-download$")) + .respond_with(ResponseTemplate::new(403).set_body_string("Forbidden")) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(RemoteCachePutResponder { + state: state.clone(), + }) + .mount(&server) + .await; + let project = remote_cache_project(&server, ""); + + assert!(run_build_with_remote_token(&project).status.success()); + remove_local_task_cache(&project); + remove_project_file(&project, "dist"); + remove_project_file(&project, "executed-marker"); + + let second = run_build_with_remote_token(&project); + + let output = format!( + "{}{}", + String::from_utf8_lossy(&second.stdout), + String::from_utf8_lossy(&second.stderr) + ); + assert!(second.status.success(), "{output}"); + assert!(project.file_exists("executed-marker"), "{output}"); + assert!( + output.contains("remote cache download returned HTTP 403"), + "{output}" + ); + assert!(!output.contains("token permissions"), "{output}"); +} + +#[tokio::test] +async fn run_remote_cache_treats_a_redirect_to_a_missing_artifact_as_a_miss() { + let server = MockServer::start().await; + let state = RemoteCacheState::default(); + mount_redirecting_remote_cache(&server, state.clone(), false).await; + let project = remote_cache_project(&server, ""); + + assert!(run_build_with_remote_token(&project).status.success()); + remove_local_task_cache(&project); + remove_project_file(&project, "dist"); + remove_project_file(&project, "executed-marker"); + + let second = run_build_with_remote_token(&project); + + assert!(second.status.success()); + assert!( + project.file_exists("executed-marker"), + "a redirect to a missing artifact must rebuild", + ); + let stderr = String::from_utf8_lossy(&second.stderr); + assert!( + !stderr.contains("remote cache download"), + "a missing artifact is a plain miss, not a failed download:\n{stderr}", + ); +} + +#[tokio::test] +async fn run_remote_cache_doesnt_fetch_storage_for_a_redirect_without_the_artifacts_hash() { + let server = MockServer::start().await; + let state = RemoteCacheState::default(); + mount_redirecting_remote_cache(&server, state.clone(), true).await; + let project = remote_cache_project(&server, ""); + + assert!(run_build_with_remote_token(&project).status.success()); + // As for an artifact Turborepo uploaded, which carries no hash. + *state.sha.lock().expect("remote cache sha mutex poisoned") = None; + remove_local_task_cache(&project); + remove_project_file(&project, "dist"); + remove_project_file(&project, "executed-marker"); + + let second = run_build_with_remote_token(&project); + + assert!(second.status.success()); + assert!( + project.file_exists("executed-marker"), + "an artifact that can't be verified must rebuild", + ); + let storage_reads = server + .received_requests() + .await + .unwrap() + .into_iter() + .filter(|request| request.url.path().starts_with("/storage/")) + .count(); + assert_eq!(storage_reads, 0); +} + +#[tokio::test] +async fn run_remote_cache_doesnt_fetch_storage_for_a_redirect_without_the_tag_a_signing_key_needs() +{ + let server = MockServer::start().await; + let state = RemoteCacheState::default(); + mount_redirecting_remote_cache(&server, state.clone(), true).await; + let project = remote_cache_project(&server, r#", "signature": true"#); + let signed_build = || { + lpm(&project) + .env("LPM_REMOTE_CACHE_TOKEN", "remote-token") + .env("LPM_REMOTE_CACHE_SIGNATURE_KEY", "signing-key") + .args(["run", "build"]) + .output() + .expect("failed to run signed remote build") + }; + + assert!(signed_build().status.success()); + *state.tag.lock().expect("remote cache tag mutex poisoned") = None; + remove_local_task_cache(&project); + remove_project_file(&project, "dist"); + remove_project_file(&project, "executed-marker"); + + let second = signed_build(); + + assert!(second.status.success()); + assert!( + project.file_exists("executed-marker"), + "an artifact without its tag must rebuild", + ); + let storage_reads = server + .received_requests() + .await + .unwrap() + .into_iter() + .filter(|request| request.url.path().starts_with("/storage/")) + .count(); + assert_eq!(storage_reads, 0); +} + #[tokio::test] async fn run_remote_cache_accepts_http_ipv6_loopback() { let listener = std::net::TcpListener::bind("[::1]:0").unwrap(); @@ -3425,6 +3774,381 @@ async fn run_remote_cache_upload_rebuilds_the_body_after_reactive_session_refres registry.server().verify().await; } +/// Two cached tasks, the second depending on the first, so one run makes two +/// remote-cache lookups and two uploads. The first task takes at least +/// `assets_ms` to build. +fn two_task_remote_cache_project(server: &MockServer, assets_ms: u64) -> TempProject { + let project = TempProject::empty(&format!( + r#"{{ + "name": "remote-cache-limits", + "version": "1.0.0", + "scripts": {{ + "assets": "node -e \"setTimeout(() => {{ const fs=require('fs'); fs.mkdirSync('assets',{{recursive:true}}); fs.writeFileSync('assets/a.txt','a') }}, {assets_ms})\"", + "build": "node -e \"const fs=require('fs'); fs.mkdirSync('dist',{{recursive:true}}); fs.writeFileSync('dist/value.txt','built')\"" + }} + }}"# + )); + project.write_file( + "lpm.json", + &format!( + r#"{{ + "remoteCache": {{ "enabled": true, "url": "{}/v8" }}, + "tasks": {{ + "assets": {{ "cache": true, "outputs": ["assets/**"] }}, + "build": {{ "cache": true, "outputs": ["dist/**"], "dependsOn": ["assets"] }} + }} + }}"#, + server.uri(), + ), + ); + project +} + +fn limit_reached(limit: &str, retry_after: u64) -> ResponseTemplate { + ResponseTemplate::new(429) + .insert_header("retry-after", retry_after.to_string().as_str()) + .set_body_json(serde_json::json!({ + "error": format!("Remote cache limit reached: {limit}"), + "retryAfter": retry_after, + })) +} + +fn successful_build_stderr(output: &std::process::Output) -> String { + let stderr = String::from_utf8_lossy(&output.stderr).into_owned(); + assert!( + output.status.success(), + "a remote cache limit must not fail the build:\n{stderr}" + ); + stderr +} + +fn parallel_remote_cache_project(server: &MockServer) -> TempProject { + let project = TempProject::empty( + r#"{ + "name": "remote-cache-parallel", + "version": "1.0.0", + "scripts": { + "one": "node -e \"const fs=require('fs'); fs.mkdirSync('one',{recursive:true}); fs.writeFileSync('one/a.txt','1')\"", + "two": "node -e \"const fs=require('fs'); fs.mkdirSync('two',{recursive:true}); fs.writeFileSync('two/a.txt','2')\"" + } + }"#, + ); + project.write_file( + "lpm.json", + &format!( + r#"{{ + "remoteCache": {{ "enabled": true, "url": "{}/v8" }}, + "tasks": {{ + "one": {{ "cache": true, "outputs": ["one/**"] }}, + "two": {{ "cache": true, "outputs": ["two/**"] }} + }} + }}"#, + server.uri(), + ), + ); + project +} + +fn run_parallel_with_remote_token(project: &TempProject) -> std::process::Output { + lpm(project) + .env("LPM_REMOTE_CACHE_TOKEN", "remote-token") + .env("LPM_REMOTE_CACHE_SIGNATURE_KEY", "signing-key") + .args(["run", "one", "two", "--parallel"]) + .output() + .expect("failed to run lpm run one two --parallel") +} + +fn refused(error: &str) -> ResponseTemplate { + ResponseTemplate::new(403) + .set_body_json(serde_json::json!({ "error": error })) + .set_delay(std::time::Duration::from_millis(400)) +} + +#[tokio::test] +async fn run_remote_cache_says_once_that_it_stops_uploads_refused_at_once() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(ResponseTemplate::new(404)) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(refused( + "Token does not have remote cache write permissions", + )) + .expect(2) + .mount(&server) + .await; + let project = parallel_remote_cache_project(&server); + + let stderr = successful_build_stderr(&run_parallel_with_remote_token(&project)); + + assert_eq!( + stderr + .matches("skipping remote cache uploads for this run") + .count(), + 1, + "two uploads refused at once must say so once:\n{stderr}", + ); + server.verify().await; +} + +#[tokio::test] +async fn run_remote_cache_says_once_that_it_stops_lookups_refused_at_once() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(refused("Token does not have remote cache read permissions")) + .expect(2) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "urls": [] }))) + .expect(0) + .mount(&server) + .await; + let project = parallel_remote_cache_project(&server); + + let stderr = successful_build_stderr(&run_parallel_with_remote_token(&project)); + + assert_eq!( + stderr + .matches("skipping the remote cache for this run") + .count(), + 1, + "two lookups refused at once must say so once:\n{stderr}", + ); + server.verify().await; +} + +#[tokio::test] +async fn run_remote_cache_skips_lookups_and_uploads_while_a_lookup_limit_lasts() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(limit_reached("2,000 cache hits a minute", 42)) + .expect(1) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "urls": [] }))) + .expect(0) + .mount(&server) + .await; + let project = two_task_remote_cache_project(&server, 0); + + let stderr = successful_build_stderr(&run_build_with_remote_token(&project)); + + assert_eq!(project.read_file("dist/value.txt"), "built"); + assert!( + stderr.contains( + "Remote cache limit reached: 2,000 cache hits a minute; skipping remote cache lookups and uploads for 42 seconds" + ), + "the warning should name the limit and the pause:\n{stderr}", + ); + server.verify().await; +} + +#[tokio::test] +async fn run_remote_cache_skips_uploads_while_an_upload_limit_lasts() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(ResponseTemplate::new(404)) + .expect(2) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(limit_reached("300 uploads a minute", 42)) + .expect(1) + .mount(&server) + .await; + let project = two_task_remote_cache_project(&server, 0); + + let stderr = successful_build_stderr(&run_build_with_remote_token(&project)); + + assert!( + stderr.contains( + "Remote cache limit reached: 300 uploads a minute; skipping remote cache uploads for 42 seconds" + ), + "the warning should name the limit and the pause:\n{stderr}", + ); + server.verify().await; +} + +#[tokio::test] +async fn run_remote_cache_looks_up_again_after_a_refusal_of_a_second_and_goes_on() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(limit_reached("2,000 cache hits a minute", 1)) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + // The refused lookup, looked up again after its second, and the second + // task's. + Mock::given(method("GET")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(ResponseTemplate::new(404)) + .expect(2) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "urls": [] }))) + .expect(2) + .mount(&server) + .await; + let project = two_task_remote_cache_project(&server, 0); + + successful_build_stderr(&run_build_with_remote_token(&project)); + + server.verify().await; +} + +#[tokio::test] +async fn run_remote_cache_warns_about_a_limit_after_another_remote_cache_warning() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(ResponseTemplate::new(500)) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(ResponseTemplate::new(404)) + .expect(1) + .mount(&server) + .await; + Mock::given(method("PUT")) + .and(path_regex(r"^/v8/artifacts/[a-fA-F0-9]+$")) + .respond_with(limit_reached("300 uploads a minute", 42)) + .expect(1) + .mount(&server) + .await; + let project = two_task_remote_cache_project(&server, 0); + + let stderr = successful_build_stderr(&run_build_with_remote_token(&project)); + + assert!( + stderr.contains("remote cache lookup returned HTTP 500"), + "{stderr}" + ); + assert!( + stderr.contains( + "Remote cache limit reached: 300 uploads a minute; skipping remote cache uploads for 42 seconds" + ), + "an earlier warning must not hide the limit's:\n{stderr}", + ); + server.verify().await; +} + +/// A cache server that advertises every artifact at `advertised_bytes` but +/// sends only a few of them, so a test can reach the CLI's size limit without +/// moving that much data. Returns the server's base URL. +fn oversized_artifact_server(advertised_bytes: u64) -> String { + use std::io::{Read as _, Write as _}; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + std::thread::spawn(move || { + for mut stream in listener.incoming().flatten() { + let mut head = Vec::new(); + let mut byte = [0u8; 1]; + while !head.ends_with(b"\r\n\r\n") && matches!(stream.read(&mut byte), Ok(1)) { + head.push(byte[0]); + } + let response = if head.starts_with(b"GET /v8/artifacts/") { + format!( + "HTTP/1.1 200 OK\r\ncontent-length: {advertised_bytes}\r\nx-artifact-sha: {}\r\nconnection: close\r\n\r\npartial", + "0".repeat(64), + ) + } else { + "HTTP/1.1 404 Not Found\r\ncontent-length: 0\r\nconnection: close\r\n\r\n".into() + }; + let _ = stream.write_all(response.as_bytes()); + } + }); + url +} + +fn run_read_only_build(project: &TempProject, registry_url: Option<&str>) -> String { + let mut command = lpm(project); + command + .env("LPM_REMOTE_CACHE_TOKEN", "remote-token") + .env("LPM_REMOTE_CACHE_READ_ONLY", "1") + .args(["run", "build"]); + match registry_url { + Some(url) => command.env("LPM_REGISTRY_URL", url), + None => command.env("LPM_REMOTE_CACHE_SIGNATURE_KEY", "signing-key"), + }; + let output = command.output().expect("failed to run lpm run build"); + let stderr = successful_build_stderr(&output); + assert_eq!(project.read_file("dist/value.txt"), "built"); + stderr +} + +fn remote_cache_project_at(url: &str) -> TempProject { + let project = TempProject::empty( + r#"{ + "name": "remote-cache-size", + "version": "1.0.0", + "scripts": { + "build": "node -e \"const fs=require('fs'); fs.mkdirSync('dist',{recursive:true}); fs.writeFileSync('dist/value.txt','built')\"" + } + }"#, + ); + project.write_file( + "lpm.json", + &format!( + r#"{{ + "remoteCache": {{ "enabled": true, "url": "{url}/v8" }}, + "tasks": {{ "build": {{ "cache": true, "outputs": ["dist/**"] }} }} + }}"# + ), + ); + project +} + +const MIB: u64 = 1024 * 1024; + +#[test] +fn run_remote_cache_skips_artifacts_over_100_mib_from_the_registry() { + let url = oversized_artifact_server(150 * MIB); + let project = remote_cache_project_at(&url); + + let stderr = run_read_only_build(&project, Some(&url)); + + assert!( + stderr.contains("remote cache artifact exceeds the 100 MiB limit; treating it as a miss"), + "{stderr}" + ); +} + +#[test] +fn run_remote_cache_reads_artifacts_up_to_500_mib_from_another_cache_server() { + let within = oversized_artifact_server(150 * MIB); + let stderr = run_read_only_build(&remote_cache_project_at(&within), None); + assert!( + stderr.contains("failed to read remote cache artifact") && !stderr.contains("exceeds"), + "a 150 MiB artifact is within another server's limit, so the CLI reads it:\n{stderr}" + ); + + let over = oversized_artifact_server(600 * MIB); + let stderr = run_read_only_build(&remote_cache_project_at(&over), None); + assert!( + stderr.contains("remote cache artifact exceeds the 500 MiB limit; treating it as a miss"), + "{stderr}" + ); +} + #[tokio::test] async fn run_remote_cache_outage_does_not_fail_successful_local_build() { let server = MockServer::start().await;