From 4fc824b6946a8082df671b2ff55d21bbd7c5f406 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 10 Sep 2026 13:58:55 +0000 Subject: [PATCH 1/5] Send POST requests once instead of replaying them on 502 A 502 comes from the proxy in front of Corgea, not from the API, so it is equally the answer for "the request never arrived" and for "the request was processed and the reply was lost coming back". Replaying every method on that status turned one `corgea scan` into a scan per attempt: both `POST /start-scan` and `POST /scan-upload` are creates, and a create the API committed before the proxy lost its answer becomes a second scan when it is sent again. Gate the replay on the method. GET and the other safe methods keep it, and so does PATCH: every PATCH the CLI sends is an upload chunk naming the byte range it fills in `Upload-Offset`, so a re-sent chunk overwrites the range it already wrote. POST is sent once and its 502 goes to the caller. `send` reads the method off the request it built; `send_with_retries` takes a closure, so its call sites now declare which method they send. Co-authored-by: Ibrahim Rahhal --- README.md | 35 +-- src/scan.rs | 44 ++-- src/utils/api.rs | 281 +++++++++++++++------- tests/cloud_commands_e2e/gateway_retry.rs | 71 +++--- 4 files changed, 273 insertions(+), 158 deletions(-) diff --git a/README.md b/README.md index 34a8ff8..f659f2d 100644 --- a/README.md +++ b/README.md @@ -48,20 +48,27 @@ 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. +### Gateway errors are retried where a second copy is harmless + +A read that the platform's proxy answers `502 Bad Gateway` replays itself, +waiting 10s, then 30s, then 50s, so a pipeline rides out the blips a busy +platform produces under parallel scans instead of failing on them. This covers +every `GET` — including the status reads a scan wait is almost entirely made of +— and the chunk uploads, which carry the byte range they fill in `Upload-Offset` +and so overwrite that range rather than appending a second copy. A request still +answered 502 after those three retries fails the command in the usual way, so a +real outage still exits non-zero. 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. + +Requests that create something — starting a scan, and uploading a report or a +source file — are sent once. A 502 comes from the proxy rather than from Corgea, +so it is equally the answer for "the request never arrived" and for "the request +was processed and the reply was lost coming back"; sending a create again does +not finish the first scan, it starts a second one. `corgea upload` treats a 502 +on a source upload as the platform being unavailable rather than one bad file +and reports the remaining paths as unsent instead of collecting the same answer +once per path. ### Skipping a re-scan of the same commit diff --git a/src/scan.rs b/src/scan.rs index 21cdd43..cd2e1c2 100644 --- a/src/scan.rs +++ b/src/scan.rs @@ -3,6 +3,7 @@ use crate::log::debug; use crate::scanners::parsers::ScanParserFactory; use crate::{utils, Config}; use reqwest::header; +use reqwest::Method; use serde_json::Value; use std::collections::HashSet; use std::io::{self, Read}; @@ -355,7 +356,7 @@ pub fn upload_scan( while attempts < 3 && !success { debug(&format!("POST: {}", src_upload_url)); - let res = utils::api::send_with_retries("a source file upload", || { + let res = utils::api::send_with_retries("a source file upload", &Method::POST, || { let form = reqwest::blocking::multipart::Form::new() .file("file", fp) .expect("Failed to read file"); @@ -373,17 +374,13 @@ 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. + // A 502 is the platform being unavailable, not something + // wrong with this one file, and an upload is a POST so + // it is not replayed. Walking the remaining paths would + // just collect the same answer once per file, 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 - ); + log::warn!("Failed to upload file {}: {}", path, status); gateway_gave_up = true; break 'files; } @@ -424,7 +421,7 @@ pub fn upload_scan( 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.", + "Stopped uploading source files: Corgea answered 502. {} of {} files were not sent.", unsent, distinct.len() ); @@ -468,15 +465,16 @@ pub fn upload_scan( index + 1, total_chunks )); - let response = utils::api::send_with_retries("a scan report chunk upload", || { - client - .post(&scan_upload_url) - .header(header::CONTENT_TYPE, "application/json") - .header("Upload-Offset", offset.to_string()) - .header("Upload-Length", input_size.to_string()) - .body(chunk.to_vec()) - .send() - }); + let response = + utils::api::send_with_retries("a scan report chunk upload", &Method::POST, || { + client + .post(&scan_upload_url) + .header(header::CONTENT_TYPE, "application/json") + .header("Upload-Offset", offset.to_string()) + .header("Upload-Length", input_size.to_string()) + .body(chunk.to_vec()) + .send() + }); let should_break = match &response { Ok(res) => { @@ -519,7 +517,7 @@ pub fn upload_scan( last_response.expect("Failed to upload scan.") } else { debug(&format!("POST: {}", scan_upload_url)); - utils::api::send_with_retries("the scan report upload", || { + utils::api::send_with_retries("the scan report upload", &Method::POST, || { client .post(&scan_upload_url) .header(header::CONTENT_TYPE, "application/json") @@ -600,7 +598,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::send_with_retries("the git config upload", || { + let res = utils::api::send_with_retries("the git config upload", &Method::POST, || { 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 95c5980..34d8a23 100644 --- a/src/utils/api.rs +++ b/src/utils/api.rs @@ -3,11 +3,11 @@ use crate::log::debug; use crate::utils; use corgea::vuln_api::{auth_header, source}; use reqwest::header::HeaderMap; -use reqwest::StatusCode; use reqwest::{ blocking::multipart, blocking::multipart::{Form, Part}, }; +use reqwest::{Method, StatusCode}; use serde::{Deserialize, Serialize}; use serde_json::json; use serde_json::Value; @@ -161,6 +161,17 @@ impl DebugRequestBuilder { None => debug(" Cookie: (none in jar for this URL)"), } + if !is_replayable(request.method()) { + debug(&format!( + " Sending once: a {} is not safe to replay.", + request.method() + )); + let response = client.execute(request)?; + debug(&format!("← {} {}", response.status(), response.url())); + debug(&format!(" Response headers: {:?}", response.headers())); + return Ok(response); + } + let mut retries = GatewayRetries::new(format!("{} {}", request.method(), request.url().path())); loop { @@ -194,6 +205,28 @@ pub fn is_gateway_error(status: StatusCode) -> bool { status == StatusCode::BAD_GATEWAY } +/// Whether a gateway error can be answered by sending the same request again. +/// +/// A 502 comes from the proxy, not from Corgea, so it says nothing about +/// whether the API handled the request: it is equally the answer for "the +/// request never arrived" and for "the request was processed and the reply was +/// lost coming back". Replaying is therefore only safe where the second copy +/// lands on top of the first, which is a property of the method: +/// +/// - `GET` and the other safe methods change nothing, and they are also the +/// volume — one scan wait is thousands of status reads against a handful of +/// writes, so this is where retrying earns its keep. +/// - `PATCH` replays. Every `PATCH` the CLI sends is an upload chunk carrying +/// the byte range it fills in `Upload-Offset`, so a re-sent chunk writes over +/// the range it already wrote instead of appending a second copy. +/// - `POST` does not replay. These are the creates: `POST /start-scan` mints a +/// new transfer and `POST /scan-upload` takes a whole report. When the API +/// commits one of those and the proxy loses the reply, sending it again does +/// not finish the first scan — it starts a second one. +fn is_replayable(method: &Method) -> bool { + method.is_safe() || method == Method::PATCH +} + /// 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. /// @@ -336,13 +369,23 @@ impl Drop for GatewayRetryGuard { /// `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. +/// +/// `method` is the one the closure sends. `send` reads the method off the +/// request it built, but a closure is opaque, so the call site has to say which +/// method it is for `is_replayable` to rule on. A method that may not be +/// replayed still gets the network-error retries, where the request never +/// reached Corgea at all. pub fn send_with_retries( operation: &str, + method: &Method, mut make_request: F, ) -> reqwest::Result where F: FnMut() -> reqwest::Result, { + if !is_replayable(method) { + return retry_on_network_error(operation, make_request); + } let _guard = GatewayRetryGuard::enter(); let mut retries = GatewayRetries::new(operation); loop { @@ -453,7 +496,7 @@ pub fn upload_zip( // 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 response_object = send_with_retries("the scan start request", &Method::POST, || { let form = reqwest::blocking::multipart::Form::new() .part( "files", @@ -535,95 +578,98 @@ pub fn upload_zip( // 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())); + let response = + match send_with_retries("a scan archive chunk upload", &Method::PATCH, || { + 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!( + Err(e) => debug(&format!( "Could not serialize the incremental file list, scanning every file: {e}" )), + } } - } - 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 - } - Err(e) => { - return Err(format!("Failed to send request: {}", e).into()); - } - }; + 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 + } + Err(e) => { + return Err(format!("Failed to send request: {}", e).into()); + } + }; if !response.status().is_success() { let status_code = response.status(); let response_text = response @@ -2756,13 +2802,57 @@ mod tests { ); } + #[test] + fn only_reads_and_offset_addressed_chunk_uploads_may_be_replayed() { + for method in [ + Method::GET, + Method::HEAD, + Method::OPTIONS, + Method::TRACE, + Method::PATCH, + ] { + assert!(is_replayable(&method), "{method} should be replayed"); + } + // A create sent twice is a second scan, so no 502 is worth a second + // send: the proxy cannot tell us whether the first one was committed. + for method in [Method::POST, Method::PUT, Method::DELETE] { + assert!(!is_replayable(&method), "{method} must be sent once"); + } + } + + #[test] + fn send_leaves_a_post_at_one_attempt() { + let (base, hits) = spawn_gateway_stub(usize::MAX); + + let response = http_client().post(&base).body("{}").send().expect("send"); + + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!( + hits.load(Ordering::SeqCst), + 1, + "a 502 on a create has to reach the caller, not be sent again" + ); + } + + #[test] + fn send_replays_a_patch() { + // The chunk uploads name the byte range they fill, so the retry that a + // create must not get is exactly the one a chunk should. + let (base, hits) = spawn_gateway_stub(2); + + let response = http_client().patch(&base).body("{}").send().expect("send"); + + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(hits.load(Ordering::SeqCst), 3, "two 502s, then the answer"); + } + #[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 response = send_with_retries("a chunk upload", &Method::PATCH, || { let form = reqwest::blocking::multipart::Form::new().text("field", "value"); - http_client().post(&base).multipart(form).send() + http_client().patch(&base).multipart(form).send() }) .expect("send"); @@ -2770,6 +2860,23 @@ mod tests { assert_eq!(hits.load(Ordering::SeqCst), 3); } + #[test] + fn send_with_retries_leaves_a_post_at_one_attempt() { + // The archive and report uploads rebuild their multipart form per + // attempt, so this is the path that turned one `corgea scan` into + // several scans; the rebuilding must not become a licence to replay. + let (base, hits) = spawn_gateway_stub(usize::MAX); + + let response = send_with_retries("a scan upload", &Method::POST, || { + let form = reqwest::blocking::multipart::Form::new().text("field", "value"); + http_client().post(&base).multipart(form).send() + }) + .expect("send"); + + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!(hits.load(Ordering::SeqCst), 1); + } + #[test] fn nested_retry_loops_do_not_multiply_the_attempts() { // A replayable body inside `send_with_retries` is retried by exactly @@ -2777,8 +2884,8 @@ mod tests { // 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() + let response = send_with_retries("a retryable body", &Method::PATCH, || { + http_client().patch(&base).body("{}").send() }) .expect("send"); diff --git a/tests/cloud_commands_e2e/gateway_retry.rs b/tests/cloud_commands_e2e/gateway_retry.rs index 2271621..839b5cc 100644 --- a/tests/cloud_commands_e2e/gateway_retry.rs +++ b/tests/cloud_commands_e2e/gateway_retry.rs @@ -1,6 +1,9 @@ //! 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. +//! pipeline, and exit non-zero only once the retries are spent — but only for +//! the requests a second copy of is harmless. A `POST` is a create, and the +//! proxy cannot say whether the API committed the first one, so those are sent +//! once and the 502 goes to the caller. //! //! 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 @@ -97,51 +100,51 @@ fn wait_exits_unclean_once_the_retries_are_spent() { } #[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(); +fn a_rejected_scan_start_is_not_sent_again() { + // The incident this guards: `POST /start-scan` mints a transfer, and the + // 502 the proxy returns is also what it returns after the API committed one + // and the reply was lost. Replaying it was turning a single `corgea scan` + // into a scan per attempt in the project. + let project = git_project(); let mut plan = vec![verify_request()]; - for _ in 0..ATTEMPTS { + for branch in ["main", "master"] { 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(), + "look up a baseline scan to diff against", + move |request| assert_baseline_lookup_request(request, "cloud-e2e", branch), + json_response(scans_response(Vec::new())), )); } + plan.push(expected_request( + "reject the scan start with a gateway error", + |request| assert_authenticated_request(request, Method::POST, "/api/v1/start-scan"), + 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", - ]); + command.args(["scan", "blast", "--project-name", "cloud-e2e"]); let output = run_with_timeout(command, &api); + // A second start-scan would be an unexpected request, and the stub fails + // the plan on it. let transcript = api.assert_finished(); let context = output_context(&output, &transcript); - assert_eq!(output.status.code(), Some(1), "{context}"); + assert_ne!(output.status.code(), Some(0), "{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}" - ); + assert!(!stderr.contains("Retrying in"), "{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. +fn a_rejected_source_upload_stops_the_whole_upload_walk() { + // A source upload is a `POST`, so the 502 is the answer rather than the + // start of a schedule. It is also the platform being unavailable rather + // than something wrong with this one file, so the paths behind it must not + // each collect the same answer: with two referenced sources, a walk that + // kept going would ask for two uploads instead of one. let project = two_source_report_project(); - let mut plan = vec![verify_request()]; - for _ in 0..ATTEMPTS { - plan.push(expected_request( + let plan = vec![ + verify_request(), + expected_request( "reject the source upload with a gateway error", |request| { assert_authenticated_request(request, Method::POST, "/api/v1/code-upload")?; @@ -154,8 +157,8 @@ fn an_exhausted_gateway_stops_the_whole_source_upload_walk() { 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); @@ -167,8 +170,8 @@ fn an_exhausted_gateway_stops_the_whole_source_upload_walk() { ]); 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. + // A second upload would be either a replay of the first file or the walk + // reaching the next one, 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}"); From 20e0bdfae193f1e5f0811e3ad62f874f400e0a2a Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 10 Sep 2026 13:58:55 +0000 Subject: [PATCH 2/5] Release 1.14.1 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 db8d97e..ca8e76b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -369,7 +369,7 @@ checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" [[package]] name = "corgea" -version = "1.14.0" +version = "1.14.1" dependencies = [ "chrono", "clap", diff --git a/Cargo.toml b/Cargo.toml index 0dd1ac7..8e8cf23 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "corgea" -version = "1.14.0" +version = "1.14.1" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html From dc702a5e1f7e52eb87c490b50f0f865d0dd4e548 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 10 Sep 2026 14:25:55 +0000 Subject: [PATCH 3/5] Say what the PATCH replay does and does not cover The offset argument covers the archive bytes: a chunk names the range it fills, so a replay overwrites rather than appends. It does not cover the transfer's completion. The chunk that fills the last of Upload-Length is the one that answers with scan_id, and an archive under CHUNK_SIZE is a single chunk, so on most repos the retried chunk is also the request that creates the scan -- which is safe only while the API keys that scan to the transfer_id already in the URL. Co-authored-by: Ibrahim Rahhal --- README.md | 4 ++++ src/utils/api.rs | 8 +++++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index f659f2d..2d1788f 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,10 @@ real outage still exits non-zero. 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 chunk that fills the last of an upload is also the one that answers with the +new scan's id, so retrying it relies on Corgea keying that scan to the transfer +it completes rather than minting one per completing request. + Requests that create something — starting a scan, and uploading a report or a source file — are sent once. A 502 comes from the proxy rather than from Corgea, so it is equally the answer for "the request never arrived" and for "the request diff --git a/src/utils/api.rs b/src/utils/api.rs index 34d8a23..83c7d17 100644 --- a/src/utils/api.rs +++ b/src/utils/api.rs @@ -218,7 +218,13 @@ pub fn is_gateway_error(status: StatusCode) -> bool { /// writes, so this is where retrying earns its keep. /// - `PATCH` replays. Every `PATCH` the CLI sends is an upload chunk carrying /// the byte range it fills in `Upload-Offset`, so a re-sent chunk writes over -/// the range it already wrote instead of appending a second copy. +/// the range it already wrote instead of appending a second copy. That covers +/// the archive bytes but not the transfer's completion: the chunk that fills +/// the last of `Upload-Length` is the one that answers with `scan_id`, and an +/// archive under `CHUNK_SIZE` is a single chunk, so on most repos this is +/// also the request that creates the scan. Replaying it is only free while +/// the API keys that scan to the `transfer_id` already in the URL instead of +/// creating one per completing request. /// - `POST` does not replay. These are the creates: `POST /start-scan` mints a /// new transfer and `POST /scan-upload` takes a whole report. When the API /// commits one of those and the proxy loses the reply, sending it again does From aef49328259014f24b29ccadd7c47b24c42d1ff1 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 10 Sep 2026 15:57:13 +0000 Subject: [PATCH 4/5] Stop replaying the archive chunk PATCH too The offset argument only covered the archive bytes. The chunk that fills the last of Upload-Length is also the one that answers with scan_id, and an archive under CHUNK_SIZE is a single chunk, so on most repos the retried chunk is the request that creates the scan -- the same exposure the POSTs were excluded for, and safe only if the API keys the scan to the transfer. Retrying is now limited to the safe methods, which leaves no replayable caller of send_with_retries: its gateway loop could never fire, and the thread-local guard existed only to stop that loop nesting inside the one in send. Both are gone, and the six upload call sites now say what they already resolved to -- retry_on_network_error. One rule in one place: send replays a read, and nothing replays a write. Co-authored-by: Ibrahim Rahhal --- README.md | 36 +-- src/scan.rs | 28 +- src/utils/api.rs | 337 ++++++++-------------- tests/cloud_commands_e2e/gateway_retry.rs | 55 ++-- 4 files changed, 173 insertions(+), 283 deletions(-) diff --git a/README.md b/README.md index 2d1788f..0ea4aa1 100644 --- a/README.md +++ b/README.md @@ -48,31 +48,27 @@ 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 where a second copy is harmless +### Gateway errors are retried on reads, never on writes A read that the platform's proxy answers `502 Bad Gateway` replays itself, waiting 10s, then 30s, then 50s, so a pipeline rides out the blips a busy -platform produces under parallel scans instead of failing on them. This covers -every `GET` — including the status reads a scan wait is almost entirely made of -— and the chunk uploads, which carry the byte range they fill in `Upload-Offset` -and so overwrite that range rather than appending a second copy. A request still -answered 502 after those three retries fails the command in the usual way, so a -real outage still exits non-zero. Each retry is logged, and the count belongs to -a single request: any successful call starts the next one with the full three +platform produces under parallel scans instead of failing on them. That covers +the status reads a scan wait is almost entirely made of. A read still answered +502 after those three retries fails the command in the usual way, so a real +outage still exits non-zero. 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 chunk that fills the last of an upload is also the one that answers with the -new scan's id, so retrying it relies on Corgea keying that scan to the transfer -it completes rather than minting one per completing request. - -Requests that create something — starting a scan, and uploading a report or a -source file — are sent once. A 502 comes from the proxy rather than from Corgea, -so it is equally the answer for "the request never arrived" and for "the request -was processed and the reply was lost coming back"; sending a create again does -not finish the first scan, it starts a second one. `corgea upload` treats a 502 -on a source upload as the platform being unavailable rather than one bad file -and reports the remaining paths as unsent instead of collecting the same answer -once per path. +Writes are sent once. A 502 comes from the proxy rather than from Corgea, so it +is equally the answer for "the request never arrived" and for "the request was +processed and the reply was lost coming back", and every write the CLI sends +creates something — a scan, a report, an uploaded source file. Sending one again +does not finish the first scan, it starts a second one. Writes keep the retries +for network errors, where nothing reached Corgea at all. + +`corgea upload` treats a 502 on a source upload as the platform being +unavailable rather than one bad file, and reports the remaining paths as unsent +instead of collecting the same answer once per path. ### Skipping a re-scan of the same commit diff --git a/src/scan.rs b/src/scan.rs index cd2e1c2..c4ca3e8 100644 --- a/src/scan.rs +++ b/src/scan.rs @@ -3,7 +3,6 @@ use crate::log::debug; use crate::scanners::parsers::ScanParserFactory; use crate::{utils, Config}; use reqwest::header; -use reqwest::Method; use serde_json::Value; use std::collections::HashSet; use std::io::{self, Read}; @@ -356,7 +355,7 @@ pub fn upload_scan( while attempts < 3 && !success { debug(&format!("POST: {}", src_upload_url)); - let res = utils::api::send_with_retries("a source file upload", &Method::POST, || { + let res = utils::api::retry_on_network_error("a source file upload", || { let form = reqwest::blocking::multipart::Form::new() .file("file", fp) .expect("Failed to read file"); @@ -375,7 +374,7 @@ pub fn upload_scan( status, body )); // A 502 is the platform being unavailable, not something - // wrong with this one file, and an upload is a POST so + // wrong with this one file, and an upload is a write so // it is not replayed. Walking the remaining paths would // just collect the same answer once per file, so stop // uploading source files altogether. @@ -465,16 +464,15 @@ pub fn upload_scan( index + 1, total_chunks )); - let response = - utils::api::send_with_retries("a scan report chunk upload", &Method::POST, || { - client - .post(&scan_upload_url) - .header(header::CONTENT_TYPE, "application/json") - .header("Upload-Offset", offset.to_string()) - .header("Upload-Length", input_size.to_string()) - .body(chunk.to_vec()) - .send() - }); + let response = utils::api::retry_on_network_error("a scan report chunk upload", || { + client + .post(&scan_upload_url) + .header(header::CONTENT_TYPE, "application/json") + .header("Upload-Offset", offset.to_string()) + .header("Upload-Length", input_size.to_string()) + .body(chunk.to_vec()) + .send() + }); let should_break = match &response { Ok(res) => { @@ -517,7 +515,7 @@ pub fn upload_scan( last_response.expect("Failed to upload scan.") } else { debug(&format!("POST: {}", scan_upload_url)); - utils::api::send_with_retries("the scan report upload", &Method::POST, || { + utils::api::retry_on_network_error("the scan report upload", || { client .post(&scan_upload_url) .header(header::CONTENT_TYPE, "application/json") @@ -598,7 +596,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::send_with_retries("the git config upload", &Method::POST, || { + let res = utils::api::retry_on_network_error("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 83c7d17..6a766ea 100644 --- a/src/utils/api.rs +++ b/src/utils/api.rs @@ -175,9 +175,7 @@ impl DebugRequestBuilder { 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`. + // Cloned before the send, which consumes the request. let replay = request.try_clone(); let response = client.execute(request)?; @@ -188,8 +186,7 @@ impl DebugRequestBuilder { 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()) - { + if !retries.wait_for_retry(response.status()) { return Ok(response); } request = replay; @@ -210,27 +207,21 @@ pub fn is_gateway_error(status: StatusCode) -> bool { /// A 502 comes from the proxy, not from Corgea, so it says nothing about /// whether the API handled the request: it is equally the answer for "the /// request never arrived" and for "the request was processed and the reply was -/// lost coming back". Replaying is therefore only safe where the second copy -/// lands on top of the first, which is a property of the method: +/// lost coming back". Only a request that changes nothing is safe to send into +/// that ambiguity, so replaying is limited to the safe methods. /// -/// - `GET` and the other safe methods change nothing, and they are also the -/// volume — one scan wait is thousands of status reads against a handful of -/// writes, so this is where retrying earns its keep. -/// - `PATCH` replays. Every `PATCH` the CLI sends is an upload chunk carrying -/// the byte range it fills in `Upload-Offset`, so a re-sent chunk writes over -/// the range it already wrote instead of appending a second copy. That covers -/// the archive bytes but not the transfer's completion: the chunk that fills -/// the last of `Upload-Length` is the one that answers with `scan_id`, and an -/// archive under `CHUNK_SIZE` is a single chunk, so on most repos this is -/// also the request that creates the scan. Replaying it is only free while -/// the API keys that scan to the `transfer_id` already in the URL instead of -/// creating one per completing request. -/// - `POST` does not replay. These are the creates: `POST /start-scan` mints a -/// new transfer and `POST /scan-upload` takes a whole report. When the API -/// commits one of those and the proxy loses the reply, sending it again does -/// not finish the first scan — it starts a second one. +/// That keeps it where it earns its keep — a scan wait is thousands of status +/// reads against a handful of writes — and off the writes, every one of which +/// creates something. `POST /start-scan` mints a transfer and `POST +/// /scan-upload` takes a whole report. The archive `PATCH` looks safer, since +/// it names the byte range it fills in `Upload-Offset` and so overwrites rather +/// than appends, but the chunk that fills the last of `Upload-Length` is also +/// the one that answers with `scan_id`, and an archive under `CHUNK_SIZE` is a +/// single chunk: on most repos that request is the one that creates the scan. +/// When the API commits any of these and the proxy loses the reply, sending it +/// again does not finish the first scan — it starts a second one. fn is_replayable(method: &Method) -> bool { - method.is_safe() || method == Method::PATCH + method.is_safe() } /// Pauses before each replay of a request the gateway rejected: a 502 has to @@ -344,64 +335,6 @@ impl GatewayRetries { } } -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. -/// -/// `method` is the one the closure sends. `send` reads the method off the -/// request it built, but a closure is opaque, so the call site has to say which -/// method it is for `is_replayable` to rule on. A method that may not be -/// replayed still gets the network-error retries, where the request never -/// reached Corgea at all. -pub fn send_with_retries( - operation: &str, - method: &Method, - mut make_request: F, -) -> reqwest::Result -where - F: FnMut() -> reqwest::Result, -{ - if !is_replayable(method) { - return retry_on_network_error(operation, make_request); - } - 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); - } - } -} - pub fn http_client() -> HttpClient { HttpClient { inner: SHARED_CLIENT.clone(), @@ -502,7 +435,7 @@ pub fn upload_zip( // 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", &Method::POST, || { + let response_object = retry_on_network_error("the scan start request", || { let form = reqwest::blocking::multipart::Form::new() .part( "files", @@ -584,98 +517,95 @@ pub fn upload_zip( // 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", &Method::PATCH, || { - 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(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())); + let response = match retry_on_network_error("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(policy) = policy.clone() { - form = form.part("target_policies", multipart::Part::text(policy.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(meta) = &metadata { - form = form.part("metadata", multipart::Part::text(meta.clone())); + if let Some(sha) = &info.sha { + form = form.part("sha", multipart::Part::text(sha.to_string())); } - // 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"), - ); - } + // 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")); } - Err(e) => debug(&format!( + } + Err(e) => debug(&format!( "Could not serialize the incremental file list, scanning every file: {e}" )), - } } + } - 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 - } - Err(e) => { - return Err(format!("Failed to send request: {}", e).into()); - } - }; + 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 + } + Err(e) => { + return Err(format!("Failed to send request: {}", e).into()); + } + }; if !response.status().is_success() { let status_code = response.status(); let response_text = response @@ -2809,19 +2739,14 @@ mod tests { } #[test] - fn only_reads_and_offset_addressed_chunk_uploads_may_be_replayed() { - for method in [ - Method::GET, - Method::HEAD, - Method::OPTIONS, - Method::TRACE, - Method::PATCH, - ] { + fn only_the_safe_methods_may_be_replayed() { + for method in [Method::GET, Method::HEAD, Method::OPTIONS, Method::TRACE] { assert!(is_replayable(&method), "{method} should be replayed"); } - // A create sent twice is a second scan, so no 502 is worth a second - // send: the proxy cannot tell us whether the first one was committed. - for method in [Method::POST, Method::PUT, Method::DELETE] { + // Every write the CLI sends creates something, and a create sent twice + // is a second scan: the proxy cannot tell us whether the first one was + // committed, so no 502 on one is worth a second send. + for method in [Method::POST, Method::PUT, Method::PATCH, Method::DELETE] { assert!(!is_replayable(&method), "{method} must be sent once"); } } @@ -2841,39 +2766,29 @@ mod tests { } #[test] - fn send_replays_a_patch() { - // The chunk uploads name the byte range they fill, so the retry that a - // create must not get is exactly the one a chunk should. - let (base, hits) = spawn_gateway_stub(2); + fn send_leaves_a_patch_at_one_attempt() { + // The archive chunk names the byte range it fills, so a replay would + // not double the bytes -- but the chunk that fills the last of them is + // also the one that answers with `scan_id`, and on an archive under + // `CHUNK_SIZE` that is the only chunk. Replaying it risks the second + // scan this whole policy exists to prevent. + let (base, hits) = spawn_gateway_stub(usize::MAX); let response = http_client().patch(&base).body("{}").send().expect("send"); - assert_eq!(response.status(), StatusCode::OK); - assert_eq!(hits.load(Ordering::SeqCst), 3, "two 502s, then the answer"); - } - - #[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 chunk upload", &Method::PATCH, || { - let form = reqwest::blocking::multipart::Form::new().text("field", "value"); - http_client().patch(&base).multipart(form).send() - }) - .expect("send"); - - assert_eq!(response.status(), StatusCode::OK); - assert_eq!(hits.load(Ordering::SeqCst), 3); + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!(hits.load(Ordering::SeqCst), 1); } #[test] - fn send_with_retries_leaves_a_post_at_one_attempt() { - // The archive and report uploads rebuild their multipart form per - // attempt, so this is the path that turned one `corgea scan` into - // several scans; the rebuilding must not become a licence to replay. + fn the_multipart_uploads_are_not_replayed_by_the_network_retries() { + // The archive and report uploads rebuild their form per attempt, since + // a multipart body is a stream with nothing to clone. That rebuilding + // is what turned one `corgea scan` into several scans, so it must not + // amount to a replay: a 502 is an answer, not a network error. let (base, hits) = spawn_gateway_stub(usize::MAX); - let response = send_with_retries("a scan upload", &Method::POST, || { + let response = retry_on_network_error("a scan upload", || { let form = reqwest::blocking::multipart::Form::new().text("field", "value"); http_client().post(&base).multipart(form).send() }) @@ -2883,22 +2798,6 @@ mod tests { assert_eq!(hits.load(Ordering::SeqCst), 1); } - #[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", &Method::PATCH, || { - http_client().patch(&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 index 839b5cc..96b75f3 100644 --- a/tests/cloud_commands_e2e/gateway_retry.rs +++ b/tests/cloud_commands_e2e/gateway_retry.rs @@ -1,9 +1,9 @@ //! 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 — but only for -//! the requests a second copy of is harmless. A `POST` is a create, and the -//! proxy cannot say whether the API committed the first one, so those are sent -//! once and the 502 goes to the caller. +//! of Corgea: replay the read on a fixed schedule instead of failing the +//! pipeline, and exit non-zero only once the retries are spent. Writes are sent +//! once — the proxy cannot say whether the API committed the first copy, and +//! every write the CLI sends creates something — so their 502 goes straight to +//! the caller. //! //! 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 @@ -186,40 +186,37 @@ fn a_rejected_source_upload_stops_the_whole_upload_walk() { } #[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. +fn a_rejected_archive_chunk_is_not_sent_again() { + // The archive chunk names the byte range it fills, so a replay would not + // double the bytes. It is excluded anyway: the chunk that fills the last of + // `Upload-Length` is the one that answers with `scan_id`, and an archive + // under the 50 MB chunk size — which this fixture, and most repos, is — has + // only that one chunk. Replaying it risks the second scan. 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. + // POST, then the chunk PATCH. Everything from the chunk on is dropped, + // since the rejected chunk ends the command. + let mut plan = blast_upload_plan(&project.sha, false, false); 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(), - ), - ); + plan.truncate(ARCHIVE_UPLOAD); + plan.push(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); + // A second chunk would be an unexpected request, and the stub fails the + // plan on it. 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}"); + assert_ne!(output.status.code(), Some(0), "{context}"); let stderr = String::from_utf8_lossy(&output.stderr); - assert!(stderr.contains("502 Bad Gateway"), "{context}"); + assert!(!stderr.contains("Retrying in"), "{context}"); } From 2dbedb3ef947684924df758a81b445911871ea84 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 10 Sep 2026 16:22:35 +0000 Subject: [PATCH 5/5] Retry 429 Too Many Requests for every method, writes included The two statuses worth retrying promise different things. A 502 comes from the proxy, so it cannot say whether the API acted, which is why a write's 502 is now the answer. A 429 is the rate limiter declining the request before the API sees it: nothing was created, so re-sending finishes the same work rather than duplicating it, and a rate-limited upload that failed the command would otherwise leave a pipeline to be re-run by hand. So the retry decision takes the method and the status together -- should_retry -- rather than testing the status alone. is_gateway_error becomes is_transient_error and covers both, which is also what the source upload walk wants: it stops early on a 502 because a write is never replayed, and on a 429 because the retries are already spent. A 429 usually names its own window, and ignoring it means spending the whole schedule inside it and failing anyway, so Retry-After is honored when present. Capped at two minutes per pause so an unverifiable header cannot stall a pipeline, and floored at the schedule so a retry cannot become a hammer. Co-authored-by: Ibrahim Rahhal --- README.md | 51 +- src/scan.rs | 44 +- src/utils/api.rs | 653 ++++++++++++------ tests/cloud_commands_e2e/main.rs | 2 +- .../{gateway_retry.rs => transient_retry.rs} | 108 ++- 5 files changed, 617 insertions(+), 241 deletions(-) rename tests/cloud_commands_e2e/{gateway_retry.rs => transient_retry.rs} (66%) diff --git a/README.md b/README.md index 0ea4aa1..7dd3d6c 100644 --- a/README.md +++ b/README.md @@ -48,27 +48,36 @@ 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 on reads, never on writes - -A read that the platform's proxy answers `502 Bad Gateway` replays itself, -waiting 10s, then 30s, then 50s, so a pipeline rides out the blips a busy -platform produces under parallel scans instead of failing on them. That covers -the status reads a scan wait is almost entirely made of. A read still answered -502 after those three retries fails the command in the usual way, so a real -outage still exits non-zero. 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. - -Writes are sent once. A 502 comes from the proxy rather than from Corgea, so it -is equally the answer for "the request never arrived" and for "the request was -processed and the reply was lost coming back", and every write the CLI sends -creates something — a scan, a report, an uploaded source file. Sending one again -does not finish the first scan, it starts a second one. Writes keep the retries -for network errors, where nothing reached Corgea at all. - -`corgea upload` treats a 502 on a source upload as the platform being -unavailable rather than one bad file, and reports the remaining paths as unsent -instead of collecting the same answer once per path. +### Rate limits and gateway errors are retried + +Neither `429 Too Many Requests` nor `502 Bad Gateway` is Corgea rejecting a +request on its merits, so both are retried rather than failed: the CLI waits +10s, then 30s, then 50s, and a pipeline rides out the blips a busy platform +produces under parallel scans instead of failing on them. A request still +answered the same way after those three retries fails the command in the usual +way, so a real outage still exits non-zero. Each retry is logged, and the count +belongs to a single request: any other answer starts the next one with the full +three retries again. A `429` that names a `Retry-After` in seconds is honored, +up to two minutes for any one pause, and never shortens the pause below the +schedule. + +Which requests get retried depends on which of the two it is: + +- A `429` retries everything, `POST` and `PATCH` included. The rate limiter + declines the request before the API sees it, so nothing was created and + sending it again finishes the same work. +- A `502` retries reads only. It comes from the proxy rather than from Corgea, + so it is equally the answer for "the request never arrived" and for "the + request was processed and the reply was lost coming back" — and every write + the CLI sends creates something, so re-sending one does not finish the first + scan, it starts a second. A write's 502 goes straight to the caller. + +Writes still retry network errors, where nothing reached Corgea at all. + +`corgea upload` treats either status on a source upload — a 502, or a rate limit +that outlived its retries — as the platform being unavailable rather than one bad +file, and reports the remaining paths as unsent instead of collecting the same +answer once per path. ### Skipping a re-scan of the same commit diff --git a/src/scan.rs b/src/scan.rs index c4ca3e8..5d486ab 100644 --- a/src/scan.rs +++ b/src/scan.rs @@ -3,6 +3,7 @@ use crate::log::debug; use crate::scanners::parsers::ScanParserFactory; use crate::{utils, Config}; use reqwest::header; +use reqwest::Method; use serde_json::Value; use std::collections::HashSet; use std::io::{self, Read}; @@ -328,7 +329,7 @@ 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; + let mut platform_declined = false; 'files: for path in &paths { if !Path::new(&path).exists() { @@ -355,7 +356,7 @@ pub fn upload_scan( while attempts < 3 && !success { debug(&format!("POST: {}", src_upload_url)); - let res = utils::api::retry_on_network_error("a source file upload", || { + let res = utils::api::send_with_retries("a source file upload", &Method::POST, || { let form = reqwest::blocking::multipart::Form::new() .file("file", fp) .expect("Failed to read file"); @@ -373,14 +374,16 @@ pub fn upload_scan( "Code upload failed with status: {}. Response body: {}", status, body )); - // A 502 is the platform being unavailable, not something - // wrong with this one file, and an upload is a write so - // it is not replayed. Walking the remaining paths would + // A 502 or a rate limit that got this far is the + // platform being unavailable, not something wrong with + // this one file: the 502 because an upload is a write + // and so is never replayed, the 429 because its retries + // are already spent. Walking the remaining paths would // just collect the same answer once per file, so stop // uploading source files altogether. - if utils::api::is_gateway_error(status) { + if utils::api::is_transient_error(status) { log::warn!("Failed to upload file {}: {}", path, status); - gateway_gave_up = true; + platform_declined = true; break 'files; } log::warn!("Failed to upload file {} {}... retrying", status, path); @@ -415,12 +418,12 @@ 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 { + if platform_declined { 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 answered 502. {} of {} files were not sent.", + "Stopped uploading source files: Corgea is not accepting them right now. {} of {} files were not sent.", unsent, distinct.len() ); @@ -464,15 +467,16 @@ pub fn upload_scan( index + 1, total_chunks )); - let response = utils::api::retry_on_network_error("a scan report chunk upload", || { - client - .post(&scan_upload_url) - .header(header::CONTENT_TYPE, "application/json") - .header("Upload-Offset", offset.to_string()) - .header("Upload-Length", input_size.to_string()) - .body(chunk.to_vec()) - .send() - }); + let response = + utils::api::send_with_retries("a scan report chunk upload", &Method::POST, || { + client + .post(&scan_upload_url) + .header(header::CONTENT_TYPE, "application/json") + .header("Upload-Offset", offset.to_string()) + .header("Upload-Length", input_size.to_string()) + .body(chunk.to_vec()) + .send() + }); let should_break = match &response { Ok(res) => { @@ -515,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("the scan report upload", || { + utils::api::send_with_retries("the scan report upload", &Method::POST, || { client .post(&scan_upload_url) .header(header::CONTENT_TYPE, "application/json") @@ -596,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("the git config upload", || { + let res = utils::api::send_with_retries("the git config upload", &Method::POST, || { 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 6a766ea..8ae3edb 100644 --- a/src/utils/api.rs +++ b/src/utils/api.rs @@ -161,19 +161,10 @@ impl DebugRequestBuilder { None => debug(" Cookie: (none in jar for this URL)"), } - if !is_replayable(request.method()) { - debug(&format!( - " Sending once: a {} is not safe to replay.", - request.method() - )); - let response = client.execute(request)?; - debug(&format!("← {} {}", response.status(), response.url())); - debug(&format!(" Response headers: {:?}", response.headers())); - return Ok(response); - } - - let mut retries = - GatewayRetries::new(format!("{} {}", request.method(), request.url().path())); + // Read off the request now: sending it consumes it, and the retry + // policy needs the method for every attempt. + let method = request.method().clone(); + let mut retries = TransientRetries::new(format!("{} {}", method, request.url().path())); loop { // Cloned before the send, which consumes the request. let replay = request.try_clone(); @@ -183,10 +174,21 @@ impl DebugRequestBuilder { debug(&format!(" Response headers: {:?}", response.headers())); let Some(replay) = replay else { - debug(" Not retrying: this request's body cannot be re-sent from here."); + // A streamed body has nothing to clone. The uploads that have + // one come through `send_with_retries`, which rebuilds the + // whole request instead of replaying this one. + if should_retry(response.status(), &method) { + debug(" Not retrying here: this request's body cannot be re-sent."); + } return Ok(response); }; - if !retries.wait_for_retry(response.status()) { + if RetryLoopGuard::outer_loop_active() + || !retries.wait_for_retry( + response.status(), + &method, + retry_after(response.headers()), + ) + { return Ok(response); } request = replay; @@ -194,55 +196,86 @@ impl DebugRequestBuilder { } } -/// 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 +/// Statuses the CLI answers with a retry rather than a failure. +/// +/// A `429` is Corgea's rate limiter turning the request away and a `502` is the +/// proxy in front of Corgea failing to get an answer from the API. Neither is +/// the API rejecting the request on its merits, so under the parallel scanning +/// load that produces them a retry succeeds where failing the pipeline would +/// not. +pub fn is_transient_error(status: StatusCode) -> bool { + matches!( + status, + StatusCode::TOO_MANY_REQUESTS | StatusCode::BAD_GATEWAY + ) } -/// Whether a gateway error can be answered by sending the same request again. +/// Whether `status` is worth sending this `method` again. /// -/// A 502 comes from the proxy, not from Corgea, so it says nothing about -/// whether the API handled the request: it is equally the answer for "the -/// request never arrived" and for "the request was processed and the reply was -/// lost coming back". Only a request that changes nothing is safe to send into -/// that ambiguity, so replaying is limited to the safe methods. +/// Both transient statuses are worth retrying, but they promise different +/// things about what happened to the request, and only one of them is safe to +/// answer by re-sending a write: /// -/// That keeps it where it earns its keep — a scan wait is thousands of status -/// reads against a handful of writes — and off the writes, every one of which -/// creates something. `POST /start-scan` mints a transfer and `POST -/// /scan-upload` takes a whole report. The archive `PATCH` looks safer, since -/// it names the byte range it fills in `Upload-Offset` and so overwrites rather -/// than appends, but the chunk that fills the last of `Upload-Length` is also -/// the one that answers with `scan_id`, and an archive under `CHUNK_SIZE` is a -/// single chunk: on most repos that request is the one that creates the scan. -/// When the API commits any of these and the proxy loses the reply, sending it -/// again does not finish the first scan — it starts a second one. -fn is_replayable(method: &Method) -> bool { - method.is_safe() -} - -/// 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. +/// - `429 Too Many Requests` is the rate limiter declining the request before +/// the API sees it. Nothing was created, so any method can be sent again — +/// which matters most for the writes, since a rate-limited upload that failed +/// the command leaves a pipeline to be re-run by hand. +/// - `502 Bad Gateway` comes from the proxy, so it says nothing about whether +/// the API acted: it is equally the answer for "never arrived" and for "was +/// processed, and the reply was lost coming back". Only a request that changes +/// nothing is safe to send into that ambiguity, and every write the CLI sends +/// creates something. `POST /start-scan` mints a transfer, `POST /scan-upload` +/// takes a whole report, and the archive `PATCH` that fills the last of +/// `Upload-Length` is the one that answers with `scan_id` — on an archive +/// under `CHUNK_SIZE`, the only chunk. Re-sending one of those does not finish +/// the first scan, it starts a second. +fn should_retry(status: StatusCode, method: &Method) -> bool { + match status { + StatusCode::TOO_MANY_REQUESTS => true, + StatusCode::BAD_GATEWAY => method.is_safe(), + _ => false, + } +} + +/// Pauses before each retry: a transient error 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] = [ +/// The schedule belongs to one request. Any other answer ends it, so the next +/// transient error — on this request or a later one — starts again at the first +/// pause. +const TRANSIENT_RETRY_DELAYS: [Duration; 3] = [ Duration::from_secs(10), Duration::from_secs(30), Duration::from_secs(50), ]; +/// The longest a single `Retry-After` may hold up one request, so a header the +/// CLI cannot sanity-check cannot stall a pipeline indefinitely. +const MAX_RETRY_AFTER: Duration = Duration::from_secs(120); + +/// The pause Corgea asked for, when it named one. +/// +/// Only the delta-seconds form is read. `Retry-After` also allows an HTTP date, +/// which would need the CLI to trust its own clock against the server's, and +/// which a rate limiter has no reason to send. +fn retry_after(headers: &HeaderMap) -> Option { + let seconds: u64 = headers + .get(reqwest::header::RETRY_AFTER)? + .to_str() + .ok()? + .trim() + .parse() + .ok()?; + Some(Duration::from_secs(seconds).min(MAX_RETRY_AFTER)) +} + /// 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(); + return TRANSIENT_RETRY_DELAYS.to_vec(); }; let parsed: Option> = raw .split(',') @@ -257,21 +290,31 @@ fn parse_retry_delays(raw: Option<&str>) -> Vec { RETRY_DELAYS_OVERRIDE_ENV, raw ); - GATEWAY_RETRY_DELAYS.to_vec() + TRANSIENT_RETRY_DELAYS.to_vec() } } } -fn gateway_retry_delays() -> &'static [Duration] { +fn transient_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() } +#[cfg(test)] +thread_local! { + /// The pauses `wait_before_retry` was asked for, so a test can read the + /// schedule's decisions off it without spending them. + static RECORDED_WAITS: std::cell::RefCell> = + const { std::cell::RefCell::new(Vec::new()) }; +} + /// Unit tests exercise the schedule, not the waiting. #[cfg(test)] -fn wait_before_retry(_delay: Duration) {} +fn wait_before_retry(delay: Duration) { + RECORDED_WAITS.with(|waits| waits.borrow_mut().push(delay)); +} #[cfg(not(test))] fn wait_before_retry(delay: Duration) { @@ -286,14 +329,14 @@ fn format_delay(delay: Duration) -> String { } } -/// One request's progress through `GATEWAY_RETRY_DELAYS`, and the log lines a +/// One request's progress through `TRANSIENT_RETRY_DELAYS`, and the log lines a /// pipeline needs to tell a retried blip from a real outage. -struct GatewayRetries { +struct TransientRetries { operation: String, retries_spent: usize, } -impl GatewayRetries { +impl TransientRetries { fn new(operation: impl Into) -> Self { Self { operation: operation.into(), @@ -303,16 +346,22 @@ impl GatewayRetries { /// 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) { + /// `false` means this response is the answer: either it is not one this + /// method may be retried for — which resets the schedule, so a later + /// transient error 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, + method: &Method, + asked_for: Option, + ) -> bool { + if !should_retry(status, method) { self.retries_spent = 0; return false; } - let delays = gateway_retry_delays(); - let Some(delay) = delays.get(self.retries_spent) else { + let delays = transient_retry_delays(); + let Some(scheduled) = delays.get(self.retries_spent) else { log::error!( "Corgea answered {} with {} on {} attempts. Giving up.", self.operation, @@ -321,20 +370,78 @@ impl GatewayRetries { ); return false; }; + // A rate limiter that names its own window knows better than the + // schedule does, and coming back sooner than it asked only spends a + // retry to be told the same thing. Never earlier than either, though: + // the schedule is the floor that keeps a retry from becoming a hammer. + let delay = asked_for.unwrap_or(*scheduled).max(*scheduled); self.retries_spent += 1; log::warn!( "Corgea answered {} with {}. Retrying in {}... ({}/{})", self.operation, status, - format_delay(*delay), + format_delay(delay), self.retries_spent, delays.len() ); - wait_before_retry(*delay); + 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_RETRY_LOOP: std::cell::Cell = const { std::cell::Cell::new(false) }; +} + +struct RetryLoopGuard(bool); + +impl RetryLoopGuard { + fn enter() -> Self { + Self(IN_RETRY_LOOP.with(|active| active.replace(true))) + } + + fn outer_loop_active() -> bool { + IN_RETRY_LOOP.with(|active| active.get()) + } +} + +impl Drop for RetryLoopGuard { + fn drop(&mut self) { + IN_RETRY_LOOP.with(|active| active.set(self.0)); + } +} + +/// Send a request that has to be rebuilt for every attempt, retrying both +/// network errors and transient statuses. +/// +/// `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. +/// +/// `method` is the one the closure sends. `send` reads the method off the +/// request it built, but a closure is opaque, so the call site has to name it +/// for `should_retry` to rule on. +pub fn send_with_retries( + operation: &str, + method: &Method, + mut make_request: F, +) -> reqwest::Result +where + F: FnMut() -> reqwest::Result, +{ + let _guard = RetryLoopGuard::enter(); + let mut retries = TransientRetries::new(operation); + loop { + let response = retry_on_network_error(operation, &mut make_request)?; + if !retries.wait_for_retry(response.status(), method, retry_after(response.headers())) { + return Ok(response); + } + } +} + pub fn http_client() -> HttpClient { HttpClient { inner: SHARED_CLIENT.clone(), @@ -435,7 +542,7 @@ pub fn upload_zip( // 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 = retry_on_network_error("the scan start request", || { + let response_object = send_with_retries("the scan start request", &Method::POST, || { let form = reqwest::blocking::multipart::Form::new() .part( "files", @@ -517,95 +624,98 @@ pub fn upload_zip( // 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 retry_on_network_error("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())); + let response = + match send_with_retries("a scan archive chunk upload", &Method::PATCH, || { + 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!( + Err(e) => debug(&format!( "Could not serialize the incremental file list, scanning every file: {e}" )), + } } - } - 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 - } - Err(e) => { - return Err(format!("Failed to send request: {}", e).into()); - } - }; + 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 + } + Err(e) => { + return Err(format!("Failed to send request: {}", e).into()); + } + }; if !response.status().is_success() { let status_code = response.status(); let response_text = response @@ -2624,6 +2734,32 @@ mod tests { /// 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) { + // What a proxy in front of the API actually returns: HTML, not the JSON + // envelope every endpoint parses. + spawn_failing_stub( + failures, + "502 Bad Gateway", + "", + "502 Bad Gateway", + ) + } + + /// Answers the first `failures` requests with 429, then 200. + fn spawn_rate_limited_stub(failures: usize) -> (String, Arc) { + spawn_failing_stub( + failures, + "429 Too Many Requests", + "", + r#"{"message":"rate limit exceeded"}"#, + ) + } + + fn spawn_failing_stub( + failures: usize, + status_line: &'static str, + extra_headers: &'static str, + body: &'static str, + ) -> (String, Arc) { use std::io::Write; let hits = Arc::new(AtomicUsize::new(0)); let counter = Arc::clone(&hits); @@ -2635,13 +2771,7 @@ mod tests { 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", - ) + corgea::vuln_api_stub::http_response(status_line, extra_headers, body) } else { corgea::vuln_api_stub::http_response("200 OK", "", r#"{"status":"ok"}"#) }; @@ -2652,17 +2782,18 @@ mod tests { } #[test] - fn gateway_retry_schedule_is_the_documented_one() { + fn transient_retry_schedule_is_the_documented_one() { // The customer contract: retry after 10s, 30s and 50s, then fail. assert_eq!( - GATEWAY_RETRY_DELAYS, + TRANSIENT_RETRY_DELAYS, [ Duration::from_secs(10), Duration::from_secs(30), Duration::from_secs(50) ] ); - assert!(is_gateway_error(StatusCode::BAD_GATEWAY)); + assert!(is_transient_error(StatusCode::BAD_GATEWAY)); + assert!(is_transient_error(StatusCode::TOO_MANY_REQUESTS)); // 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. @@ -2673,33 +2804,13 @@ mod tests { StatusCode::INTERNAL_SERVER_ERROR, StatusCode::SERVICE_UNAVAILABLE, ] { - assert!(!is_gateway_error(status), "{status}"); + assert!(!is_transient_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(None), TRANSIENT_RETRY_DELAYS.to_vec()); assert_eq!( parse_retry_delays(Some("50, 100")), vec![Duration::from_millis(50), Duration::from_millis(100)] @@ -2708,12 +2819,37 @@ mod tests { for raw in ["", "abc", "50,,100", "1.5"] { assert_eq!( parse_retry_delays(Some(raw)), - GATEWAY_RETRY_DELAYS.to_vec(), + TRANSIENT_RETRY_DELAYS.to_vec(), "{raw:?}" ); } } + #[test] + fn retry_after_reads_only_a_whole_number_of_seconds_and_is_capped() { + let header = |value: &str| { + let mut headers = HeaderMap::new(); + headers.insert(reqwest::header::RETRY_AFTER, value.parse().unwrap()); + headers + }; + assert_eq!( + retry_after(&header("30")), + Some(Duration::from_secs(30)), + "the delta-seconds form is what a rate limiter sends" + ); + assert_eq!( + retry_after(&header("99999")), + Some(MAX_RETRY_AFTER), + "one header must not be able to stall a pipeline indefinitely" + ); + assert_eq!(retry_after(&HeaderMap::new()), None); + // The HTTP-date form and anything malformed fall back to the schedule + // rather than being read as a pause of zero. + for raw in ["Wed, 21 Oct 2026 07:28:00 GMT", "", "soon", "1.5", "-5"] { + assert_eq!(retry_after(&header(raw)), None, "{raw:?}"); + } + } + #[test] fn send_retries_bad_gateway_until_the_server_answers() { let (base, hits) = spawn_gateway_stub(2); @@ -2733,26 +2869,106 @@ mod tests { assert_eq!(response.status(), StatusCode::BAD_GATEWAY); assert_eq!( hits.load(Ordering::SeqCst), - GATEWAY_RETRY_DELAYS.len() + 1, + TRANSIENT_RETRY_DELAYS.len() + 1, "the caller must see the 502 once the retries are spent" ); } #[test] - fn only_the_safe_methods_may_be_replayed() { + fn a_rate_limit_is_retried_for_every_method_and_a_gateway_error_only_for_reads() { + // 429 is the rate limiter declining the request before the API sees it, + // so nothing was created and re-sending finishes the same work. + for method in [ + Method::GET, + Method::HEAD, + Method::POST, + Method::PUT, + Method::PATCH, + Method::DELETE, + ] { + assert!( + should_retry(StatusCode::TOO_MANY_REQUESTS, &method), + "{method} should be retried after a rate limit" + ); + } + // 502 comes from the proxy, which cannot say whether the API acted, and + // every write the CLI sends creates something. for method in [Method::GET, Method::HEAD, Method::OPTIONS, Method::TRACE] { - assert!(is_replayable(&method), "{method} should be replayed"); + assert!(should_retry(StatusCode::BAD_GATEWAY, &method), "{method}"); } - // Every write the CLI sends creates something, and a create sent twice - // is a second scan: the proxy cannot tell us whether the first one was - // committed, so no 502 on one is worth a second send. for method in [Method::POST, Method::PUT, Method::PATCH, Method::DELETE] { - assert!(!is_replayable(&method), "{method} must be sent once"); + assert!( + !should_retry(StatusCode::BAD_GATEWAY, &method), + "{method} must be sent once" + ); + } + // Nothing else is retried for any method: the API answering for itself + // is an answer. + for status in [ + StatusCode::OK, + StatusCode::UNAUTHORIZED, + StatusCode::INTERNAL_SERVER_ERROR, + StatusCode::SERVICE_UNAVAILABLE, + ] { + assert!(!should_retry(status, &Method::GET), "{status}"); + assert!(!should_retry(status, &Method::POST), "{status}"); } } #[test] - fn send_leaves_a_post_at_one_attempt() { + fn the_schedule_stops_after_its_retries_and_any_other_answer_resets_it() { + let mut retries = TransientRetries::new("GET /api/v1/scan/s1"); + for _ in 0..TRANSIENT_RETRY_DELAYS.len() { + assert!(retries.wait_for_retry(StatusCode::BAD_GATEWAY, &Method::GET, None)); + } + assert!( + !retries.wait_for_retry(StatusCode::BAD_GATEWAY, &Method::GET, None), + "the fourth 502 is the answer, not a fourth retry" + ); + + // Any other answer 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, &Method::GET, None)); + for _ in 0..TRANSIENT_RETRY_DELAYS.len() { + assert!(retries.wait_for_retry(StatusCode::TOO_MANY_REQUESTS, &Method::GET, None)); + } + assert!(!retries.wait_for_retry(StatusCode::TOO_MANY_REQUESTS, &Method::GET, None)); + } + + /// The pauses the schedule has asked for on this thread, forgetting them. + fn drain_recorded_waits() -> Vec { + RECORDED_WAITS.with(|waits| std::mem::take(&mut *waits.borrow_mut())) + } + + #[test] + fn a_retry_after_lengthens_the_pause_but_never_shortens_it() { + let asked_for = TRANSIENT_RETRY_DELAYS[0] + Duration::from_secs(5); + let mut retries = TransientRetries::new("POST /api/v1/scan-upload"); + drain_recorded_waits(); + + let rate_limited = |retries: &mut TransientRetries, header| { + assert!(retries.wait_for_retry(StatusCode::TOO_MANY_REQUESTS, &Method::POST, header)); + }; + // A window the limiter named, longer than the schedule knows about. + rate_limited(&mut retries, Some(asked_for)); + // "Come back immediately", which the schedule floor overrides: a retry + // must not become a hammer. + rate_limited(&mut retries, Some(Duration::ZERO)); + // And no header at all, which is the schedule as written. + rate_limited(&mut retries, None); + + assert_eq!( + drain_recorded_waits(), + vec![ + asked_for, + TRANSIENT_RETRY_DELAYS[1], + TRANSIENT_RETRY_DELAYS[2] + ] + ); + } + + #[test] + fn send_leaves_a_post_at_one_attempt_on_a_gateway_error() { let (base, hits) = spawn_gateway_stub(usize::MAX); let response = http_client().post(&base).body("{}").send().expect("send"); @@ -2766,7 +2982,7 @@ mod tests { } #[test] - fn send_leaves_a_patch_at_one_attempt() { + fn send_leaves_a_patch_at_one_attempt_on_a_gateway_error() { // The archive chunk names the byte range it fills, so a replay would // not double the bytes -- but the chunk that fills the last of them is // also the one that answers with `scan_id`, and on an archive under @@ -2781,14 +2997,46 @@ mod tests { } #[test] - fn the_multipart_uploads_are_not_replayed_by_the_network_retries() { - // The archive and report uploads rebuild their form per attempt, since - // a multipart body is a stream with nothing to clone. That rebuilding - // is what turned one `corgea scan` into several scans, so it must not - // amount to a replay: a 502 is an answer, not a network error. + fn send_retries_a_rate_limited_post() { + let (base, hits) = spawn_rate_limited_stub(2); + + let response = http_client().post(&base).body("{}").send().expect("send"); + + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + hits.load(Ordering::SeqCst), + 3, + "two rate limits, then the upload goes through" + ); + } + + #[test] + fn send_with_retries_rebuilds_a_rate_limited_multipart_upload() { + // The archive and report uploads have streamed bodies with nothing to + // clone, so `send` cannot replay them and the call sites rebuild the + // form here instead. A rate limit is the case that needs it: the upload + // never reached the API, so the CLI has to send it again or leave the + // pipeline to be re-run by hand. + let (base, hits) = spawn_rate_limited_stub(2); + + let response = send_with_retries("a scan upload", &Method::POST, || { + 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 send_with_retries_leaves_a_multipart_upload_at_one_attempt_on_a_gateway_error() { + // Rebuilding the form per attempt must not amount to a licence to + // replay: this is the path that turned one `corgea scan` into several + // scans. let (base, hits) = spawn_gateway_stub(usize::MAX); - let response = retry_on_network_error("a scan upload", || { + let response = send_with_retries("a scan upload", &Method::POST, || { let form = reqwest::blocking::multipart::Form::new().text("field", "value"); http_client().post(&base).multipart(form).send() }) @@ -2798,6 +3046,25 @@ mod tests { assert_eq!(hits.load(Ordering::SeqCst), 1); } + #[test] + fn nested_retry_loops_do_not_multiply_the_attempts() { + // A body `send` can clone, inside `send_with_retries`: the two loops + // would otherwise each run the schedule, which is sixteen requests and + // six minutes of waiting for one upload. + let (base, hits) = spawn_rate_limited_stub(usize::MAX); + + let response = send_with_retries("a retryable body", &Method::POST, || { + http_client().post(&base).body("{}").send() + }) + .expect("send"); + + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!( + hits.load(Ordering::SeqCst), + TRANSIENT_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/main.rs b/tests/cloud_commands_e2e/main.rs index b5b2b5c..6d4cc51 100644 --- a/tests/cloud_commands_e2e/main.rs +++ b/tests/cloud_commands_e2e/main.rs @@ -3,9 +3,9 @@ mod repo_common; mod block_on_report; mod common; -mod gateway_retry; mod inspect; mod scan_incremental; mod scan_list; mod scan_skip; +mod transient_retry; mod upload_wait; diff --git a/tests/cloud_commands_e2e/gateway_retry.rs b/tests/cloud_commands_e2e/transient_retry.rs similarity index 66% rename from tests/cloud_commands_e2e/gateway_retry.rs rename to tests/cloud_commands_e2e/transient_retry.rs index 96b75f3..91dd585 100644 --- a/tests/cloud_commands_e2e/gateway_retry.rs +++ b/tests/cloud_commands_e2e/transient_retry.rs @@ -1,9 +1,12 @@ -//! The CLI's answer to intermittent `502 Bad Gateway` from the proxy in front -//! of Corgea: replay the read on a fixed schedule instead of failing the -//! pipeline, and exit non-zero only once the retries are spent. Writes are sent -//! once — the proxy cannot say whether the API committed the first copy, and -//! every write the CLI sends creates something — so their 502 goes straight to -//! the caller. +//! The CLI's answer to the two statuses that are not Corgea rejecting a request +//! on its merits: retry on a fixed schedule instead of failing the pipeline, and +//! exit non-zero only once the retries are spent. +//! +//! Which requests get that retry depends on which status it is. A `429` is the +//! rate limiter declining the request before the API sees it, so every method is +//! sent again, writes included. A `502` comes from the proxy, which cannot say +//! whether the API acted, so only reads are replayed and a write's 502 goes +//! straight to the caller. //! //! 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 @@ -29,6 +32,13 @@ fn bad_gateway() -> (StatusCode, String) { ) } +fn too_many_requests() -> (StatusCode, String) { + ( + StatusCode::TOO_MANY_REQUESTS, + r#"{"message":"rate limit exceeded"}"#.to_string(), + ) +} + fn rejected_scan_read(scan_id: &'static str) -> ExpectedRequest { let path = format!("/api/v1/scan/{scan_id}"); expected_request( @@ -99,6 +109,92 @@ fn wait_exits_unclean_once_the_retries_are_spent() { ); } +#[test] +fn a_rate_limited_scan_start_goes_through_on_a_retry() { + // The upload bodies are streamed multipart forms, which cannot be replayed + // from a request that was already built — so this also proves the form is + // rebuilt per attempt, since the planned start-scan 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 this rate-limits twice before letting through, then the chunk PATCH. + const SCAN_START: usize = 3; + for _ in 0..2 { + plan.insert( + SCAN_START, + expected_request( + "rate-limit the scan start", + |request| assert_authenticated_request(request, Method::POST, "/api/v1/start-scan"), + too_many_requests(), + ), + ); + } + 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); + // A rate limit never reached the API, so re-sending the create finishes the + // one scan rather than starting another. + 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("429 Too Many Requests"), "{context}"); + assert!(stderr.contains("Retrying in"), "{context}"); + assert!(!stderr.contains("Giving up"), "{context}"); +} + +#[test] +fn an_exhausted_rate_limit_stops_the_whole_source_upload_walk() { + // The retries are spent on the platform declining to take uploads, not on + // one file, so the paths behind it must not each spend the schedule again: + // with two referenced sources that would be 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( + "rate-limit the source upload", + |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(()) + }, + too_many_requests(), + )); + } + 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); + assert!(stderr.contains("Giving up"), "{context}"); + // 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}"); +} + #[test] fn a_rejected_scan_start_is_not_sent_again() { // The incident this guards: `POST /start-scan` mints a transfer, and the