diff --git a/Cargo.lock b/Cargo.lock index ef4fc73..db8d97e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -369,7 +369,7 @@ checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" [[package]] name = "corgea" -version = "1.13.0" +version = "1.14.0" dependencies = [ "chrono", "clap", diff --git a/Cargo.toml b/Cargo.toml index 32467ac..0dd1ac7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "corgea" -version = "1.13.0" +version = "1.14.0" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html diff --git a/README.md b/README.md index d075b7d..34a8ff8 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,8 @@ third-party report. All three exit 1 if the scan fails, printing the reason and the scanners that hit problems. A scan that completes with a scanner missing exits 0 with a warning. -Waiting gives up after 10 hours; override with `CORGEA_SCAN_TIMEOUT_SECONDS`. +While waiting, the scan's status is read every 3 seconds. Waiting gives up after +10 hours; override with `CORGEA_SCAN_TIMEOUT_SECONDS`. `--fail`/`--block-on` then wait up to 15 minutes for blocking rules to be evaluated; override with `CORGEA_BLOCKING_RULES_TIMEOUT_SECONDS`. @@ -47,6 +48,21 @@ evaluated; override with `CORGEA_BLOCKING_RULES_TIMEOUT_SECONDS`. trips: both are written before `--fail`/`--block-on` are evaluated, so a scan that exits 1 on a blocking rule still leaves its report behind to ingest. +### Gateway errors are retried, not surfaced + +Every Corgea API call the CLI makes — uploads included — replays itself when the +platform's proxy answers `502 Bad Gateway`, waiting 10s, then 30s, then 50s. A +request that is still answered 502 after those three retries fails the command +in the usual way, so a pipeline exits non-zero on a real outage and rides out the +blips a busy platform produces under parallel scans. Each retry is logged, and +the count belongs to a single request: any successful call starts the next one +with the full three retries again. + +The one place this stops early is `corgea upload`'s per-file source upload: a 502 +that outlives the retries there is taken as the platform being unavailable rather +than one bad file, so the remaining paths are reported as unsent instead of each +spending another 90 seconds. + ### Skipping a re-scan of the same commit A pipeline that re-runs on an unchanged commit can reuse the scan it already diff --git a/src/scan.rs b/src/scan.rs index 8bafa1e..21cdd43 100644 --- a/src/scan.rs +++ b/src/scan.rs @@ -328,8 +328,9 @@ pub fn upload_scan( let mut uploaded_paths = HashSet::new(); let mut uploaded_count = 0; let mut upload_error_count = 0; + let mut gateway_gave_up = false; - for path in &paths { + 'files: for path in &paths { if !Path::new(&path).exists() { log::error!( "Required file {} not found which is required for the scan, exiting.", @@ -354,7 +355,7 @@ pub fn upload_scan( while attempts < 3 && !success { debug(&format!("POST: {}", src_upload_url)); - let res = utils::api::retry_on_network_error("file upload", || { + let res = utils::api::send_with_retries("a source file upload", || { let form = reqwest::blocking::multipart::Form::new() .file("file", fp) .expect("Failed to read file"); @@ -372,6 +373,20 @@ pub fn upload_scan( "Code upload failed with status: {}. Response body: {}", status, body )); + // A 502 that outlived the retry schedule is the platform + // being unavailable, not this one file. Retrying it here + // would spend the schedule twice over, and walking the + // remaining paths would spend a fresh 90 seconds on each + // of them, so stop uploading source files altogether. + if utils::api::is_gateway_error(status) { + log::warn!( + "Failed to upload file {} after the gateway retries: {}", + path, + status + ); + gateway_gave_up = true; + break 'files; + } log::warn!("Failed to upload file {} {}... retrying", status, path); std::thread::sleep(std::time::Duration::from_secs(1)); attempts += 1; @@ -402,6 +417,19 @@ pub fn upload_scan( } } + // Everything the aborted walk never attempted still counts as unsent, or + // the closing summary would report one failure for a whole skipped tree. + if gateway_gave_up { + let distinct: HashSet<&String> = paths.iter().collect(); + let unsent = distinct.len() - uploaded_paths.len(); + upload_error_count += unsent; + log::warn!( + "Stopped uploading source files: Corgea was still answering 502 after the retries. {} of {} files were not sent.", + unsent, + distinct.len() + ); + } + if uploaded_count == 0 { log::error!("Failed to upload any files for the scan, exiting."); std::process::exit(1); @@ -440,7 +468,7 @@ pub fn upload_scan( index + 1, total_chunks )); - let response = utils::api::retry_on_network_error("scan chunk upload", || { + let response = utils::api::send_with_retries("a scan report chunk upload", || { client .post(&scan_upload_url) .header(header::CONTENT_TYPE, "application/json") @@ -491,7 +519,7 @@ pub fn upload_scan( last_response.expect("Failed to upload scan.") } else { debug(&format!("POST: {}", scan_upload_url)); - utils::api::retry_on_network_error("scan upload", || { + utils::api::send_with_retries("the scan report upload", || { client .post(&scan_upload_url) .header(header::CONTENT_TYPE, "application/json") @@ -572,7 +600,7 @@ pub fn upload_scan( if git_config_path.exists() { debug("Uploading .git/config"); debug(&format!("POST: {}", git_config_upload_url)); - let res = utils::api::retry_on_network_error("git config upload", || { + let res = utils::api::send_with_retries("the git config upload", || { let form = reqwest::blocking::multipart::Form::new() .file("file", git_config_path) .expect("Failed to read file"); diff --git a/src/scanners/blast.rs b/src/scanners/blast.rs index ca813fe..e8082a2 100644 --- a/src/scanners/blast.rs +++ b/src/scanners/blast.rs @@ -16,6 +16,15 @@ use std::time::{Duration, Instant}; const SCAN_TIMEOUT_ENV: &str = "CORGEA_SCAN_TIMEOUT_SECONDS"; const DEFAULT_SCAN_TIMEOUT: Duration = Duration::from_secs(10 * 60 * 60); +/// How long to pause between scan status reads. +/// +/// Scans run for minutes, so reading more often than this only adds load: at +/// one read a second, a single wait against the default 10-hour budget can +/// reach 36,000 requests, and a pipeline scanning in parallel multiplies that +/// by every concurrent wait. Three seconds costs a pipeline no more than a few +/// seconds of extra latency on the final status. +const SCAN_POLL_INTERVAL: Duration = Duration::from_secs(3); + /// Overrides how long the CI gate waits for blocking rules to be evaluated. const BLOCKING_RULES_TIMEOUT_ENV: &str = "CORGEA_BLOCKING_RULES_TIMEOUT_SECONDS"; @@ -1034,7 +1043,7 @@ pub fn wait_for_scan(config: &Config, scan_id: &str, budget: WaitBudget) { let mut last_status = String::from("unknown"); let result = loop { - thread::sleep(Duration::from_secs(1)); + thread::sleep(SCAN_POLL_INTERVAL); // Every read is capped to what is left of the budget: on the client's // own timeout a stalled read would otherwise keep us going long past // the wait the user asked for. @@ -1781,6 +1790,18 @@ mod tests { assert_eq!(format_timeout(Duration::from_secs(90)), "90s"); } + #[test] + fn poll_interval_is_the_documented_one_and_short_against_the_budget() { + assert_eq!(SCAN_POLL_INTERVAL, Duration::from_secs(3)); + // A pause that ever grew toward the budget would spend the whole wait + // asleep and report a finished scan long after it finished. + assert!( + SCAN_POLL_INTERVAL < DEFAULT_SCAN_TIMEOUT / 100, + "poll interval {SCAN_POLL_INTERVAL:?} is not short against the \ + {DEFAULT_SCAN_TIMEOUT:?} budget it runs inside" + ); + } + #[test] fn blocking_rules_timeout_outlasts_the_doghouse_wait_windows() { // This side fails closed on its own deadline, so it must never expire diff --git a/src/utils/api.rs b/src/utils/api.rs index f02db74..95c5980 100644 --- a/src/utils/api.rs +++ b/src/utils/api.rs @@ -140,14 +140,16 @@ impl DebugRequestBuilder { pub fn send(self) -> reqwest::Result { use reqwest::cookie::CookieStore; + let DebugRequestBuilder { client, inner } = self; + let token = AUTH_TOKEN.read().unwrap().clone(); let builder = if !token.is_empty() { - self.inner.headers(auth_headers(&token)) + inner.headers(auth_headers(&token)) } else { - self.inner + inner }; - let request = builder.build()?; + let mut request = builder.build()?; debug(&format!("→ {} {}", request.method(), request.url())); debug(&format!(" Request headers: {:?}", request.headers())); @@ -159,12 +161,195 @@ impl DebugRequestBuilder { None => debug(" Cookie: (none in jar for this URL)"), } - let response = self.client.execute(request)?; + let mut retries = + GatewayRetries::new(format!("{} {}", request.method(), request.url().path())); + loop { + // Cloned before the send, which consumes the request. A streamed + // body (the multipart uploads) has nothing to clone, so those call + // sites rebuild the whole request through `send_with_retries`. + let replay = request.try_clone(); + let response = client.execute(request)?; + + debug(&format!("← {} {}", response.status(), response.url())); + debug(&format!(" Response headers: {:?}", response.headers())); + + let Some(replay) = replay else { + debug(" Not retrying: this request's body cannot be re-sent from here."); + return Ok(response); + }; + if GatewayRetryGuard::outer_loop_active() || !retries.wait_for_retry(response.status()) + { + return Ok(response); + } + request = replay; + } + } +} + +/// Statuses worth replaying the same request for. A 502 is the proxy in front +/// of Corgea reporting it could not get an answer from the API, so the request +/// itself is usually still good — under the parallel scanning load that +/// produces these, the retry succeeds where failing the pipeline would not. +pub fn is_gateway_error(status: StatusCode) -> bool { + status == StatusCode::BAD_GATEWAY +} + +/// Pauses before each replay of a request the gateway rejected: a 502 has to +/// survive four attempts spread over 90 seconds before a command fails. +/// +/// The schedule belongs to one request. Any answer that is not a gateway error +/// ends it, so the next 502 — on this request or a later one — starts again at +/// the first pause. +const GATEWAY_RETRY_DELAYS: [Duration; 3] = [ + Duration::from_secs(10), + Duration::from_secs(30), + Duration::from_secs(50), +]; + +/// Overrides the pauses above with a comma-separated list of milliseconds, so +/// the retry path can be exercised end to end without spending 90 seconds. +const RETRY_DELAYS_OVERRIDE_ENV: &str = "DEBUG_CORGEA_OVERRIDE_RETRY_DELAYS_MS"; + +fn parse_retry_delays(raw: Option<&str>) -> Vec { + let Some(raw) = raw else { + return GATEWAY_RETRY_DELAYS.to_vec(); + }; + let parsed: Option> = raw + .split(',') + .map(|part| part.trim().parse::().ok().map(Duration::from_millis)) + .collect(); + match parsed.filter(|delays| !delays.is_empty()) { + Some(delays) => delays, + // A bad override must not silently take the retries away. + None => { + log::warn!( + "Ignoring {}='{}': expected a comma-separated list of whole milliseconds. Retrying on the default schedule instead.", + RETRY_DELAYS_OVERRIDE_ENV, + raw + ); + GATEWAY_RETRY_DELAYS.to_vec() + } + } +} + +fn gateway_retry_delays() -> &'static [Duration] { + static DELAYS: std::sync::LazyLock> = std::sync::LazyLock::new(|| { + parse_retry_delays(std::env::var(RETRY_DELAYS_OVERRIDE_ENV).ok().as_deref()) + }); + DELAYS.as_slice() +} - debug(&format!("← {} {}", response.status(), response.url())); - debug(&format!(" Response headers: {:?}", response.headers())); +/// Unit tests exercise the schedule, not the waiting. +#[cfg(test)] +fn wait_before_retry(_delay: Duration) {} - Ok(response) +#[cfg(not(test))] +fn wait_before_retry(delay: Duration) { + std::thread::sleep(delay); +} + +fn format_delay(delay: Duration) -> String { + if delay >= Duration::from_secs(1) { + format!("{}s", delay.as_secs()) + } else { + format!("{}ms", delay.as_millis()) + } +} + +/// One request's progress through `GATEWAY_RETRY_DELAYS`, and the log lines a +/// pipeline needs to tell a retried blip from a real outage. +struct GatewayRetries { + operation: String, + retries_spent: usize, +} + +impl GatewayRetries { + fn new(operation: impl Into) -> Self { + Self { + operation: operation.into(), + retries_spent: 0, + } + } + + /// Whether to send `operation` again, having waited for the next pause. + /// + /// `false` means this response is the answer: either it is not a gateway + /// error — which resets the schedule, so a later 502 gets the full run of + /// retries again — or the retries are spent and the caller should fail. + fn wait_for_retry(&mut self, status: StatusCode) -> bool { + if !is_gateway_error(status) { + self.retries_spent = 0; + return false; + } + let delays = gateway_retry_delays(); + let Some(delay) = delays.get(self.retries_spent) else { + log::error!( + "Corgea answered {} with {} on {} attempts. Giving up.", + self.operation, + status, + delays.len() + 1 + ); + return false; + }; + self.retries_spent += 1; + log::warn!( + "Corgea answered {} with {}. Retrying in {}... ({}/{})", + self.operation, + status, + format_delay(*delay), + self.retries_spent, + delays.len() + ); + wait_before_retry(*delay); + true + } +} + +thread_local! { + /// Set while `send_with_retries` is running the schedule itself, so `send` + /// does not stack a second one underneath it — four attempts each would be + /// sixteen requests and six minutes of waiting. + static IN_GATEWAY_RETRY_LOOP: std::cell::Cell = const { std::cell::Cell::new(false) }; +} + +struct GatewayRetryGuard(bool); + +impl GatewayRetryGuard { + fn enter() -> Self { + Self(IN_GATEWAY_RETRY_LOOP.with(|active| active.replace(true))) + } + + fn outer_loop_active() -> bool { + IN_GATEWAY_RETRY_LOOP.with(|active| active.get()) + } +} + +impl Drop for GatewayRetryGuard { + fn drop(&mut self) { + IN_GATEWAY_RETRY_LOOP.with(|active| active.set(self.0)); + } +} + +/// Send a request that has to be rebuilt for every attempt, retrying both +/// network errors and gateway errors. +/// +/// `send` replays a request it can clone; a multipart body is a stream with +/// nothing to clone, so the upload call sites come through here and build a +/// fresh form each time. `operation` names the request in the retry logs. +pub fn send_with_retries( + operation: &str, + mut make_request: F, +) -> reqwest::Result +where + F: FnMut() -> reqwest::Result, +{ + let _guard = GatewayRetryGuard::enter(); + let mut retries = GatewayRetries::new(operation); + loop { + let response = retry_on_network_error(operation, &mut make_request)?; + if !retries.wait_for_retry(response.status()) { + return Ok(response); + } } } @@ -266,18 +451,22 @@ pub fn upload_zip( "file_size": file_size }); - let form = reqwest::blocking::multipart::Form::new() - .part( - "files", - reqwest::blocking::multipart::Part::bytes(Vec::new()).file_name(file_name.to_string()), - ) - .text("json", json_object.to_string()); - - let response_object = client - .post(format!("{}{}/start-scan", url, API_BASE)) - .query(&[("scan_type", "blast")]) - .multipart(form) - .send(); + // The form is built per attempt: a multipart body is a stream, so a retry + // has nothing to replay unless the whole request is made again. + let response_object = send_with_retries("the scan start request", || { + let form = reqwest::blocking::multipart::Form::new() + .part( + "files", + reqwest::blocking::multipart::Part::bytes(Vec::new()) + .file_name(file_name.to_string()), + ) + .text("json", json_object.to_string()); + client + .post(format!("{}{}/start-scan", url, API_BASE)) + .query(&[("scan_type", "blast")]) + .multipart(form) + .send() + }); let response_object = match response_object { Ok(response) => { check_for_warnings(response.headers(), response.status()); @@ -344,86 +533,89 @@ pub fn upload_zip( let chunk = &buffer[..bytes_read]; - let mut form = Form::new() - .part( - "chunk_data", - Part::bytes(chunk.to_vec()) - .file_name(file_name.to_string()) - .mime_str("application/octet-stream")?, - ) - .part( - "project_name", - multipart::Part::text(project_name.to_string()), - ) - .part("file_size", multipart::Part::text(file_size.to_string())); - if let Some(ref info) = repo_info { - if let Some(branch) = &info.branch { - form = form.part("branch", multipart::Part::text(branch.to_string())); + // Rebuilt per attempt: a multipart body is a stream, so a retry has + // nothing to replay unless the whole request is made again. + let response = match send_with_retries("a scan archive chunk upload", || { + let mut form = Form::new() + .part( + "chunk_data", + Part::bytes(chunk.to_vec()) + .file_name(file_name.to_string()) + .mime_str("application/octet-stream")?, + ) + .part( + "project_name", + multipart::Part::text(project_name.to_string()), + ) + .part("file_size", multipart::Part::text(file_size.to_string())); + if let Some(ref info) = repo_info { + if let Some(branch) = &info.branch { + form = form.part("branch", multipart::Part::text(branch.to_string())); + } + if let Some(repo_url) = &info.repo_url { + form = form.part("repo_url", multipart::Part::text(repo_url.to_string())); + } + if let Some(sha) = &info.sha { + form = form.part("sha", multipart::Part::text(sha.to_string())); + } + // Always send: omitted = old CLI; false = clean HEAD snapshot. + form = form.part( + "dirty", + multipart::Part::text(if info.dirty { DIRTY_TRUE } else { DIRTY_FALSE }), + ); } - if let Some(repo_url) = &info.repo_url { - form = form.part("repo_url", multipart::Part::text(repo_url.to_string())); + if let Some(scan_type) = scan_type.clone() { + let scan_type = if scan_type.contains("blast") { + "base".to_string() + } else { + scan_type + }; + form = form.part("scan_configs", multipart::Part::text(scan_type.to_string())); } - if let Some(sha) = &info.sha { - form = form.part("sha", multipart::Part::text(sha.to_string())); + if let Some(policy) = policy.clone() { + form = form.part("target_policies", multipart::Part::text(policy.to_string())); } - // Always send: omitted = old CLI; false = clean HEAD snapshot. - form = form.part( - "dirty", - multipart::Part::text(if info.dirty { DIRTY_TRUE } else { DIRTY_FALSE }), - ); - } - if let Some(scan_type) = scan_type.clone() { - let scan_type = if scan_type.contains("blast") { - "base".to_string() - } else { - scan_type - }; - form = form.part("scan_configs", multipart::Part::text(scan_type.to_string())); - } - if let Some(policy) = policy.clone() { - form = form.part("target_policies", multipart::Part::text(policy.to_string())); - } - if let Some(meta) = &metadata { - form = form.part("metadata", multipart::Part::text(meta.clone())); - } - // Both fields or neither: the list is only safe next to the commit it - // was measured from, and a server seeing one without the other would - // guess a baseline. A list that will not serialize drops both, leaving - // a full scan. - if let Some(plan) = &incremental { - match serde_json::to_string(&plan.changed_files) { - Ok(changed_files) => { - form = form.part( - "incremental_base_sha", - multipart::Part::text(plan.base_sha.clone()), - ); - form = form.part( - "incremental_changed_files", - multipart::Part::text(changed_files), - ); - // Tells the server the list describes the working tree, not - // just a commit range, which is the only way it can accept a - // diff from a dirty upload. - if plan.covers_worktree { - form = - form.part("incremental_covers_worktree", multipart::Part::text("true")); + if let Some(meta) = &metadata { + form = form.part("metadata", multipart::Part::text(meta.clone())); + } + // Both fields or neither: the list is only safe next to the commit + // it was measured from, and a server seeing one without the other + // would guess a baseline. A list that will not serialize drops + // both, leaving a full scan. + if let Some(plan) = &incremental { + match serde_json::to_string(&plan.changed_files) { + Ok(changed_files) => { + form = form.part( + "incremental_base_sha", + multipart::Part::text(plan.base_sha.clone()), + ); + form = form.part( + "incremental_changed_files", + multipart::Part::text(changed_files), + ); + // Tells the server the list describes the working tree, + // not just a commit range, which is the only way it can + // accept a diff from a dirty upload. + if plan.covers_worktree { + form = form + .part("incremental_covers_worktree", multipart::Part::text("true")); + } } + Err(e) => debug(&format!( + "Could not serialize the incremental file list, scanning every file: {e}" + )), } - Err(e) => debug(&format!( - "Could not serialize the incremental file list, scanning every file: {e}" - )), } - } - let response = match client - .patch(format!("{}{}/start-scan/{}/", url, API_BASE, transfer_id)) - .header("Upload-Offset", offset.to_string()) - .header("Upload-Length", file_size.to_string()) - .header("Upload-Name", file_name) - .query(&[("scan_type", "blast")]) - .multipart(form) - .send() - { + client + .patch(format!("{}{}/start-scan/{}/", url, API_BASE, transfer_id)) + .header("Upload-Offset", offset.to_string()) + .header("Upload-Length", file_size.to_string()) + .header("Upload-Name", file_name) + .query(&[("scan_type", "blast")]) + .multipart(form) + .send() + }) { Ok(response) => { check_for_warnings(response.headers(), response.status()); response @@ -561,6 +753,18 @@ pub fn get_scan_issues( } Err(e) => return Err(format!("Failed to send request: {}", e).into()), }; + // A 5xx body is the gateway's or the server's, not the API's JSON, and + // reading it as a parse failure hides the status a pipeline needs. 4xx + // still goes through the body, which carries `no_project_found`. + let status = response.status(); + if status.is_server_error() { + let body = response.text().unwrap_or_default(); + debug(&format!( + "Issue listing failed: HTTP {}. Response body: {}", + status, body + )); + return Err(format!("Request failed with status: {}", status).into()); + } let response_text = response.text()?; let project_issues_response: ProjectIssuesResponse = serde_json::from_str(&response_text) .map_err(|e| { @@ -2045,6 +2249,8 @@ mod tests { use std::cell::Cell; use std::net::TcpListener; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; use std::thread; use std::time::Duration; @@ -2433,6 +2639,153 @@ mod tests { .is_err()); } + /// Answers the first `failures` requests with 502, then 200. Returns the + /// base URL and the request counter. + fn spawn_gateway_stub(failures: usize) -> (String, Arc) { + use std::io::Write; + let hits = Arc::new(AtomicUsize::new(0)); + let counter = Arc::clone(&hits); + let listener = TcpListener::bind("127.0.0.1:0").expect("bind stub"); + let base = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port()); + thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { continue }; + let _ = corgea::vuln_api_stub::read_http_request(&mut stream); + let served = counter.fetch_add(1, Ordering::SeqCst); + let response = if served < failures { + // What a proxy in front of the API actually returns: HTML, + // not the JSON envelope every endpoint parses. + corgea::vuln_api_stub::http_response( + "502 Bad Gateway", + "", + "502 Bad Gateway", + ) + } else { + corgea::vuln_api_stub::http_response("200 OK", "", r#"{"status":"ok"}"#) + }; + let _ = stream.write_all(response.as_bytes()); + } + }); + (base, hits) + } + + #[test] + fn gateway_retry_schedule_is_the_documented_one() { + // The customer contract: retry after 10s, 30s and 50s, then fail. + assert_eq!( + GATEWAY_RETRY_DELAYS, + [ + Duration::from_secs(10), + Duration::from_secs(30), + Duration::from_secs(50) + ] + ); + assert!(is_gateway_error(StatusCode::BAD_GATEWAY)); + // Everything else is the API answering for itself, including the other + // 5xx: those are not what the pipelines are hitting, and replaying an + // upload the server did read is not free. + for status in [ + StatusCode::OK, + StatusCode::UNAUTHORIZED, + StatusCode::NOT_FOUND, + StatusCode::INTERNAL_SERVER_ERROR, + StatusCode::SERVICE_UNAVAILABLE, + ] { + assert!(!is_gateway_error(status), "{status}"); + } + } + + #[test] + fn gateway_retries_stop_after_the_schedule_and_a_success_resets_them() { + let mut retries = GatewayRetries::new("GET /api/v1/scan/s1"); + for _ in 0..GATEWAY_RETRY_DELAYS.len() { + assert!(retries.wait_for_retry(StatusCode::BAD_GATEWAY)); + } + assert!( + !retries.wait_for_retry(StatusCode::BAD_GATEWAY), + "the fourth 502 is the answer, not a fourth retry" + ); + + // Any answer that is not a gateway error puts the full schedule back, + // so a later blip is retried rather than counted against the last one. + assert!(!retries.wait_for_retry(StatusCode::OK)); + for _ in 0..GATEWAY_RETRY_DELAYS.len() { + assert!(retries.wait_for_retry(StatusCode::BAD_GATEWAY)); + } + assert!(!retries.wait_for_retry(StatusCode::BAD_GATEWAY)); + } + + #[test] + fn retry_delay_override_falls_back_to_the_schedule_when_unusable() { + assert_eq!(parse_retry_delays(None), GATEWAY_RETRY_DELAYS.to_vec()); + assert_eq!( + parse_retry_delays(Some("50, 100")), + vec![Duration::from_millis(50), Duration::from_millis(100)] + ); + // A typo must not quietly leave the CLI without retries. + for raw in ["", "abc", "50,,100", "1.5"] { + assert_eq!( + parse_retry_delays(Some(raw)), + GATEWAY_RETRY_DELAYS.to_vec(), + "{raw:?}" + ); + } + } + + #[test] + fn send_retries_bad_gateway_until_the_server_answers() { + let (base, hits) = spawn_gateway_stub(2); + + let response = http_client().get(&base).send().expect("send"); + + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(hits.load(Ordering::SeqCst), 3, "two 502s, then the answer"); + } + + #[test] + fn send_gives_up_after_the_last_retry() { + let (base, hits) = spawn_gateway_stub(usize::MAX); + + let response = http_client().get(&base).send().expect("send"); + + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!( + hits.load(Ordering::SeqCst), + GATEWAY_RETRY_DELAYS.len() + 1, + "the caller must see the 502 once the retries are spent" + ); + } + + #[test] + fn send_with_retries_replays_a_multipart_body_send_cannot_clone() { + let (base, hits) = spawn_gateway_stub(2); + + let response = send_with_retries("a multipart upload", || { + let form = reqwest::blocking::multipart::Form::new().text("field", "value"); + http_client().post(&base).multipart(form).send() + }) + .expect("send"); + + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(hits.load(Ordering::SeqCst), 3); + } + + #[test] + fn nested_retry_loops_do_not_multiply_the_attempts() { + // A replayable body inside `send_with_retries` is retried by exactly + // one of the two loops: four attempts each would be sixteen requests + // and six minutes of waiting. + let (base, hits) = spawn_gateway_stub(usize::MAX); + + let response = send_with_retries("a retryable body", || { + http_client().post(&base).body("{}").send() + }) + .expect("send"); + + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!(hits.load(Ordering::SeqCst), GATEWAY_RETRY_DELAYS.len() + 1); + } + #[test] fn retry_on_network_error_gives_up_after_max_retries() { let attempts = Cell::new(0); diff --git a/tests/cli_scan_wait_terminal_state.rs b/tests/cli_scan_wait_terminal_state.rs index 2ad0989..4fb3acc 100644 --- a/tests/cli_scan_wait_terminal_state.rs +++ b/tests/cli_scan_wait_terminal_state.rs @@ -53,8 +53,19 @@ fn spawn_scan_api( reason: &'static str, errors: &'static str, ) -> String { + spawn_counted_scan_api(statuses, reason, errors).0 +} + +/// As `spawn_scan_api`, and also hands back the scan-read counter, so a test +/// whose point is that the wait kept polling can prove the reads happened. +fn spawn_counted_scan_api( + statuses: &'static [&'static str], + reason: &'static str, + errors: &'static str, +) -> (String, Arc) { let reads = Arc::new(AtomicUsize::new(0)); - common::spawn_http_stub(move |path| { + let counter = Arc::clone(&reads); + let url = common::spawn_http_stub(move |path| { if path.contains("/issues") { return ("200 OK", issues_json()); } @@ -64,7 +75,8 @@ fn spawn_scan_api( return ("200 OK", scan_json(status, reason, errors)); } ("200 OK", String::from(r#"{"status":"ok"}"#)) - }) + }); + (url, counter) } /// Run `corgea wait ` against `url`, failing rather than blocking @@ -229,16 +241,26 @@ fn already_completed_scan_reports_missing_scanner_results() { #[test] fn wait_stops_polling_a_scan_that_never_finishes() { // Guards the timeout: without it, a scan stuck in a non-terminal status - // polls forever. - let url = spawn_scan_api(&["processing"], "", ""); + // polls forever. The budget has to span several poll intervals, or the wait + // would expire before making a single poll and prove nothing. + let (url, reads) = spawn_counted_scan_api(&["processing"], "", ""); - let (code, output) = run_wait(&url, &[("CORGEA_SCAN_TIMEOUT_SECONDS", "3")]); + let (code, output) = run_wait(&url, &[("CORGEA_SCAN_TIMEOUT_SECONDS", "7")]); assert_eq!(code, Some(1), "a timeout must fail the command: {output}"); assert!( output.contains("Stopped waiting"), "timeout must explain itself: {output}" ); + // The initial read plus at least one poll. A poll interval that outgrew the + // budget would time out without polling and pass this test for the wrong + // reason. + assert!( + reads.load(Ordering::SeqCst) > 1, + "the wait polled {} times on a 7s budget, so it never exercised the \ + poll loop: {output}", + reads.load(Ordering::SeqCst) + ); } #[test] @@ -273,7 +295,10 @@ fn wait_honors_the_timeout_when_a_status_read_stalls() { // The budget has to bound the whole wait, not just the gaps between reads. // Each read carries the client's 150s timeout, so a server that accepts the // connection and never answers used to hold the CLI far past the budget. + // The budget must outlast one poll interval, or the stalling read below is + // never reached. let reads = Arc::new(AtomicUsize::new(0)); + let counter = Arc::clone(&reads); let url = common::spawn_http_stub(move |path| { if path.starts_with("/api/v1/scan/") { if reads.fetch_add(1, Ordering::SeqCst) > 0 { @@ -287,7 +312,7 @@ fn wait_honors_the_timeout_when_a_status_read_stalls() { }); let started = Instant::now(); - let (code, output) = run_wait(&url, &[("CORGEA_SCAN_TIMEOUT_SECONDS", "3")]); + let (code, output) = run_wait(&url, &[("CORGEA_SCAN_TIMEOUT_SECONDS", "5")]); let elapsed = started.elapsed(); assert_eq!(code, Some(1), "a timeout must fail the command: {output}"); @@ -297,6 +322,12 @@ fn wait_honors_the_timeout_when_a_status_read_stalls() { ); assert!( elapsed < Duration::from_secs(15), - "waited {elapsed:?} on a 3s budget: {output}" + "waited {elapsed:?} on a 5s budget: {output}" + ); + // The stalling read is the second one, so a budget that expired during the + // first poll interval would never reach the behavior under test. + assert!( + counter.load(Ordering::SeqCst) > 1, + "the stalling read was never reached: {output}" ); } diff --git a/tests/cloud_commands_e2e/common/mod.rs b/tests/cloud_commands_e2e/common/mod.rs index b9548d3..35bbf70 100644 --- a/tests/cloud_commands_e2e/common/mod.rs +++ b/tests/cloud_commands_e2e/common/mod.rs @@ -537,6 +537,24 @@ pub(crate) fn report_project() -> ReportProject { ReportProject { root, report_path } } +/// A report referencing two source files, so a test can tell "one upload +/// stopped" from "every upload was attempted". +pub(crate) fn two_source_report_project() -> ReportProject { + let root = TempDir::new().expect("create report project"); + let source_dir = root.path().join("src"); + std::fs::create_dir(&source_dir).expect("create source directory"); + for name in ["main.py", "helper.py"] { + std::fs::write(source_dir.join(name), SOURCE_BODY).expect("write source"); + } + let report_path = root.path().join("semgrep.json"); + std::fs::write( + &report_path, + r#"{"version":"semgrep.dev/v1","results":[{"path":"src/main.py"},{"path":"src/helper.py"}]}"#, + ) + .expect("write report"); + ReportProject { root, report_path } +} + pub(crate) fn git_project() -> GitProject { let root = TempDir::new().expect("create Git project"); for args in [ diff --git a/tests/cloud_commands_e2e/gateway_retry.rs b/tests/cloud_commands_e2e/gateway_retry.rs new file mode 100644 index 0000000..2271621 --- /dev/null +++ b/tests/cloud_commands_e2e/gateway_retry.rs @@ -0,0 +1,222 @@ +//! The CLI's answer to intermittent `502 Bad Gateway` from the proxy in front +//! of Corgea: replay the request on a fixed schedule instead of failing the +//! pipeline, and exit non-zero only once the retries are spent. +//! +//! The stub's plan is ordered and rejects unexpected requests, so these tests +//! pin the exact attempt count as well as the outcome — a retry loop that runs +//! twice over (or not at all) fails the plan rather than passing quietly. + +use crate::common::*; +use hyper::{Method, StatusCode}; + +/// The pauses are 10s/30s/50s in production; a run under test cannot spend 90 +/// seconds, so it retries on the same schedule compressed to milliseconds. +const FAST_RETRIES: (&str, &str) = ("DEBUG_CORGEA_OVERRIDE_RETRY_DELAYS_MS", "50,50,50"); + +/// The retry budget: three replays, so four attempts in all. +const ATTEMPTS: usize = 4; + +/// What a gateway returns when it cannot reach the API: HTML, not the JSON +/// envelope the endpoints parse. +fn bad_gateway() -> (StatusCode, String) { + ( + StatusCode::BAD_GATEWAY, + "502 Bad Gateway502 Bad Gateway" + .to_string(), + ) +} + +fn rejected_scan_read(scan_id: &'static str) -> ExpectedRequest { + let path = format!("/api/v1/scan/{scan_id}"); + expected_request( + "reject the scan read with a gateway error", + move |request| assert_authenticated_request(request, Method::GET, &path), + bad_gateway(), + ) +} + +#[test] +fn wait_rides_out_a_gateway_blip_on_the_scan_read() { + let project = tempfile::TempDir::new().expect("create wait project"); + let local_project = temp_project_name(project.path()); + let scan_id = "flaky-gateway-scan"; + let mut plan = vec![verify_request()]; + plan.push(rejected_scan_read(scan_id)); + plan.push(rejected_scan_read(scan_id)); + append_wait_plan(&mut plan, &local_project, scan_id, &["complete"]); + let api = ApiStub::start(plan); + let (mut command, _home) = cloud_command(&api, project.path()); + command.env(FAST_RETRIES.0, FAST_RETRIES.1); + command.args(["wait", scan_id]); + + let output = run_with_timeout(command, &api); + // Every planned request was made and none beyond them: the two 502s were + // replayed, and the scan read that followed was not. + let transcript = api.assert_finished(); + let context = output_context(&output, &transcript); + assert_eq!(output.status.code(), Some(0), "{context}"); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + stdout.contains("Scan has been processed successfully!"), + "{context}" + ); + // The retries are reported, so a pipeline log shows a ridden-out blip + // rather than an unexplained pause. + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("502 Bad Gateway"), "{context}"); + assert!(stderr.contains("Retrying in"), "{context}"); + assert!(!stderr.contains("Giving up"), "{context}"); +} + +#[test] +fn wait_exits_unclean_once_the_retries_are_spent() { + let project = tempfile::TempDir::new().expect("create wait project"); + let scan_id = "down-gateway-scan"; + let mut plan = vec![verify_request()]; + for _ in 0..ATTEMPTS { + plan.push(rejected_scan_read(scan_id)); + } + let api = ApiStub::start(plan); + let (mut command, _home) = cloud_command(&api, project.path()); + command.env(FAST_RETRIES.0, FAST_RETRIES.1); + command.args(["wait", scan_id]); + + let output = run_with_timeout(command, &api); + // Exactly four attempts: a fifth would be an unexpected request, and a + // third would leave one planned request unserved. + let transcript = api.assert_finished(); + let context = output_context(&output, &transcript); + assert_eq!(output.status.code(), Some(1), "{context}"); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("502 Bad Gateway"), "{context}"); + assert!(stderr.contains("Giving up"), "{context}"); + assert!( + stderr.contains(&format!("Unable to read scan '{scan_id}'")), + "{context}" + ); +} + +#[test] +fn a_source_upload_is_not_retried_past_the_schedule() { + // `corgea upload` retries a failed source upload three times of its own + // accord. Those attempts must not each spend the gateway schedule again: + // that is 12 requests and four and a half minutes for one file. + let project = report_project(); + let mut plan = vec![verify_request()]; + for _ in 0..ATTEMPTS { + plan.push(expected_request( + "reject the source upload with a gateway error", + |request| assert_authenticated_request(request, Method::POST, "/api/v1/code-upload"), + bad_gateway(), + )); + } + let api = ApiStub::start(plan); + let (mut command, _home) = cloud_command(&api, project.path()); + command.env(FAST_RETRIES.0, FAST_RETRIES.1); + command.args([ + "upload", + project.report_path().to_str().expect("UTF-8 report path"), + "--project-name", + "upload-contract", + ]); + + let output = run_with_timeout(command, &api); + let transcript = api.assert_finished(); + let context = output_context(&output, &transcript); + assert_eq!(output.status.code(), Some(1), "{context}"); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("Giving up"), "{context}"); + assert!( + stderr.contains("Failed to upload any files for the scan"), + "{context}" + ); +} + +#[test] +fn an_exhausted_gateway_stops_the_whole_source_upload_walk() { + // The schedule is spent on the platform being unavailable, not on one file, + // so the paths behind the failed one must not each start a fresh 90s of + // retries. With two referenced sources, a walk that kept going would ask + // for eight uploads instead of four. + let project = two_source_report_project(); + let mut plan = vec![verify_request()]; + for _ in 0..ATTEMPTS { + plan.push(expected_request( + "reject the source upload with a gateway error", + |request| { + assert_authenticated_request(request, Method::POST, "/api/v1/code-upload")?; + // Whichever of the two sources is walked first: the point is + // how many uploads are attempted, not their order. + let path = query_value(request, "path")?; + if !path.starts_with("src/") { + return Err(format!("unexpected upload path {path}")); + } + Ok(()) + }, + bad_gateway(), + )); + } + let api = ApiStub::start(plan); + let (mut command, _home) = cloud_command(&api, project.path()); + command.env(FAST_RETRIES.0, FAST_RETRIES.1); + command.args([ + "upload", + project.report_path().to_str().expect("UTF-8 report path"), + "--project-name", + "upload-contract", + ]); + + let output = run_with_timeout(command, &api); + // A fifth upload would be the second file starting the schedule again, and + // the stub rejects it as unexpected. + let transcript = api.assert_finished(); + let context = output_context(&output, &transcript); + assert_eq!(output.status.code(), Some(1), "{context}"); + let stderr = String::from_utf8_lossy(&output.stderr); + // The file that was never attempted is still reported as unsent, so the + // summary cannot read as a single bad file. + assert!(stderr.contains("2 of 2 files were not sent"), "{context}"); + assert!( + stderr.contains("Failed to upload any files for the scan"), + "{context}" + ); +} + +#[test] +fn blast_upload_replays_an_archive_chunk_the_gateway_rejects() { + // The upload bodies are streamed multipart forms, which cannot be replayed + // from a built request — this is what proves the form is rebuilt, since the + // planned chunk request that follows asserts every field of it. + let project = git_project(); + let mut plan = blast_upload_plan(&project.sha, false, false); + // `blast_upload_plan` order: verify, two baseline lookups, the start-scan + // POST, then the chunk PATCH this rejects once before letting it through. + const ARCHIVE_UPLOAD: usize = 4; + plan.insert( + ARCHIVE_UPLOAD, + expected_request( + "reject the archive chunk with a gateway error", + |request| { + assert_authenticated_request( + request, + Method::PATCH, + "/api/v1/start-scan/transfer-123/", + ) + }, + bad_gateway(), + ), + ); + let api = ApiStub::start(plan); + let (mut command, _home) = cloud_command(&api, project.path()); + command.env(FAST_RETRIES.0, FAST_RETRIES.1); + command.args(["scan", "blast", "--project-name", "cloud-e2e"]); + + let output = run_with_timeout(command, &api); + let transcript = api.assert_finished(); + let context = output_context(&output, &transcript); + assert_eq!(output.status.code(), Some(0), "{context}"); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!(stdout.contains("Scan Completed Successfully"), "{context}"); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("502 Bad Gateway"), "{context}"); +} diff --git a/tests/cloud_commands_e2e/main.rs b/tests/cloud_commands_e2e/main.rs index 439f10a..b5b2b5c 100644 --- a/tests/cloud_commands_e2e/main.rs +++ b/tests/cloud_commands_e2e/main.rs @@ -3,6 +3,7 @@ mod repo_common; mod block_on_report; mod common; +mod gateway_retry; mod inspect; mod scan_incremental; mod scan_list; diff --git a/tests/common/mod.rs b/tests/common/mod.rs index b807a55..d250df4 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -52,6 +52,7 @@ pub fn corgea_isolated() -> (Command, TempDir) { .env_remove("GITHUB_PR") .env_remove("REPO_DATA") .env_remove("DEBUG_CORGEA_OVERRIDE_REPORT_CHUNK_SIZE") + .env_remove("DEBUG_CORGEA_OVERRIDE_RETRY_DELAYS_MS") .env_remove("RUST_LOG") .env_remove("CORGEA_DEBUG") .env_remove("HTTP_PROXY")