From f564d039d87fb33d05237be6048f90823512f716 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 9 Sep 2026 14:13:07 +0000 Subject: [PATCH 1/4] Retry 502 Bad Gateway on every API call instead of failing the pipeline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A pipeline running many parallel scans intermittently gets 502 from the proxy in front of Corgea, and the CLI turned each one straight into a failed command — sometimes as an opaque parse error, since a gateway's HTML body is not the JSON envelope the endpoints read. Wrapping the CLI in shell retries is worse: it cannot tell a load blip from a scan that genuinely failed. Requests now replay themselves when the answer is 502, waiting 10s, then 30s, then 50s. A request still answered 502 on the fourth attempt fails the command as before, so a real outage still exits non-zero. The count belongs to one request: any successful call starts the next one with the full three retries. The retry lives in the API service's send path, so it covers every endpoint. Streamed multipart bodies (the archive and report uploads) cannot be replayed from a built request, so those call sites go through `send_with_retries`, which rebuilds the form per attempt; a thread-local guard keeps the two layers from stacking into 16 attempts. The source upload's own three-attempt loop stops on a gateway error rather than spending the schedule again per file. Also reads the status before the body on the issue listing, so an exhausted 502 reports the status instead of a JSON parse failure. Co-authored-by: Ibrahim Rahhal --- README.md | 10 + src/scan.rs | 20 +- src/utils/api.rs | 537 ++++++++++++++++++---- tests/cloud_commands_e2e/gateway_retry.rs | 172 +++++++ tests/cloud_commands_e2e/main.rs | 1 + tests/common/mod.rs | 1 + 6 files changed, 645 insertions(+), 96 deletions(-) create mode 100644 tests/cloud_commands_e2e/gateway_retry.rs diff --git a/README.md b/README.md index d075b7d..b806876 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,16 @@ 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. + ### 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..aba86b9 100644 --- a/src/scan.rs +++ b/src/scan.rs @@ -354,7 +354,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 +372,18 @@ pub fn upload_scan( "Code upload failed with status: {}. Response body: {}", status, body )); + // A gateway error already spent this request's retry + // schedule; three more one-second attempts would spend + // it again for every file. + if utils::api::is_gateway_error(status) { + upload_error_count += 1; + log::warn!( + "Failed to upload file {} after the gateway retries: {}. skipping...", + path, + status + ); + break; + } log::warn!("Failed to upload file {} {}... retrying", status, path); std::thread::sleep(std::time::Duration::from_secs(1)); attempts += 1; @@ -440,7 +452,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 +503,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 +584,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/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/cloud_commands_e2e/gateway_retry.rs b/tests/cloud_commands_e2e/gateway_retry.rs new file mode 100644 index 0000000..bdc4cae --- /dev/null +++ b/tests/cloud_commands_e2e/gateway_retry.rs @@ -0,0 +1,172 @@ +//! 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 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") From b45b422b503ed1599edef03183b5c2e5e4e30ca7 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 9 Sep 2026 14:19:29 +0000 Subject: [PATCH 2/4] Bump version from 1.13.0 to 1.14.0 Co-authored-by: Ibrahim Rahhal --- Cargo.lock | 2 +- Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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 From 17fdb7d3dc5e775297173e743e3cf88219fc2c70 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 9 Sep 2026 14:48:56 +0000 Subject: [PATCH 3/4] Poll scan status every 3 seconds instead of every second Co-authored-by: Ibrahim Rahhal --- README.md | 3 +- src/scanners/blast.rs | 23 +++++++++++++- tests/cli_scan_wait_terminal_state.rs | 45 ++++++++++++++++++++++----- 3 files changed, 62 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index b806876..f0d4d0c 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`. 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/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}" ); } From 6cff32b7a2f6d531ca125dcc045154091bdf057b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 9 Sep 2026 15:01:07 +0000 Subject: [PATCH 4/4] Stop the source upload walk once the gateway retries are spent Co-authored-by: Ibrahim Rahhal --- README.md | 5 +++ src/scan.rs | 30 ++++++++++---- tests/cloud_commands_e2e/common/mod.rs | 18 ++++++++ tests/cloud_commands_e2e/gateway_retry.rs | 50 +++++++++++++++++++++++ 4 files changed, 96 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index f0d4d0c..34a8ff8 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,11 @@ 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 aba86b9..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.", @@ -372,17 +373,19 @@ pub fn upload_scan( "Code upload failed with status: {}. Response body: {}", status, body )); - // A gateway error already spent this request's retry - // schedule; three more one-second attempts would spend - // it again for every file. + // 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) { - upload_error_count += 1; log::warn!( - "Failed to upload file {} after the gateway retries: {}. skipping...", + "Failed to upload file {} after the gateway retries: {}", path, status ); - break; + gateway_gave_up = true; + break 'files; } log::warn!("Failed to upload file {} {}... retrying", status, path); std::thread::sleep(std::time::Duration::from_secs(1)); @@ -414,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); 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 index bdc4cae..2271621 100644 --- a/tests/cloud_commands_e2e/gateway_retry.rs +++ b/tests/cloud_commands_e2e/gateway_retry.rs @@ -132,6 +132,56 @@ fn a_source_upload_is_not_retried_past_the_schedule() { ); } +#[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