diff --git a/src/api/fossa.rs b/src/api/fossa.rs index 557db17e..767fe97a 100644 --- a/src/api/fossa.rs +++ b/src/api/fossa.rs @@ -317,7 +317,7 @@ pub async fn upload_scan( opts: &Config, project: &ProjectMetadata, cli: &CliMetadata, - source_units: SourceUnits, + source_units: &SourceUnits, ) -> Result { let url = opts.endpoint().join("api/builds/custom")?; @@ -360,7 +360,7 @@ pub async fn upload_scan( run_request::(req) .await - .change_context_lazy(|| Error::upload_scan(&locator, &source_units))? + .change_context_lazy(|| Error::upload_scan(&locator, source_units))? .into() } diff --git a/src/debug.rs b/src/debug.rs index 5cbcdd40..bce3889d 100644 --- a/src/debug.rs +++ b/src/debug.rs @@ -128,7 +128,7 @@ impl Config { tracing_subscriber::fmt::layer() .compact() .with_file(false) - .with_level(false) + .with_level(true) .with_line_number(false) .with_target(false) .with_writer(std::io::stderr) diff --git a/src/subcommand/run.rs b/src/subcommand/run.rs index 97c1754f..0ebe355d 100644 --- a/src/subcommand/run.rs +++ b/src/subcommand/run.rs @@ -3,9 +3,9 @@ use std::sync::Arc; use std::time::Duration; -use error_stack::{Result, ResultExt}; -use futures::TryStreamExt; +use error_stack::{Context, Result, ResultExt}; use futures::{future::try_join_all, try_join, StreamExt}; +use futures::{Future, TryStreamExt}; use indoc::indoc; use serde::{Deserialize, Serialize}; use tap::TapFallible; @@ -203,33 +203,21 @@ async fn poll_integration( let remote = integration.remote().to_owned(); let poll_interval = integration.poll_interval().as_duration(); - // [`Retry`] needs a function that runs without any arguments to perform the retry, so turn the method into a closure. - let get_references = || async { - match integration.references().await { - Ok(success) => Ok(success), - Err(err) => { - warn!("attempt to poll integration at {remote} failed: {err:#}"); - Err(err) - } - } - }; - loop { info!("Polling '{integration}'"); - // Given that this operation is not latency sensitive, and temporary network issues can interfere, - // retry several times before permanently failing since a permanent failure means Broker shuts down - // entirely. - let strategy = ExponentialBackoff::from_millis(1000).map(jitter).take(10); - let references = Retry::spawn(strategy, get_references) - .await - .change_context(Error::PollIntegration) - .describe_lazy(|| format!("poll for changes at {remote} in integration: {integration}")) - .help(indoc! {" - Issues with this process are usually related to network errors, but may be due to misconfiguration. - Each time this polling operation was attempted, it logged a warning; please review those - warnings in the logs for more details. - "})?; + let references = retry_default(format!("poll '{integration}'"), || async { + match integration.references().await { + Ok(success) => Ok(success), + Err(err) => { + warn!("attempt to poll integration at {remote} failed: {err:#}"); + Err(err) + } + } + }) + .await + .change_context(Error::PollIntegration) + .describe_lazy(|| format!("poll for changes at {remote} in integration: {integration}"))?; // Filter to the list of references that are new since we last saw them. let references = futures::stream::iter(references.into_iter()) @@ -310,11 +298,14 @@ async fn scan_git_references( mut receiver: Receiver, mut uploader: Sender, ) -> Result<(), Error> { - let cli = fossa_cli::find_or_download( - &ctx.app, - ctx.config.debug().location(), - DesiredVersion::Latest, - ) + let cli = retry_default("download_fossa_cli", || async { + fossa_cli::find_or_download( + &ctx.app, + ctx.config.debug().location(), + DesiredVersion::Latest, + ) + .await + }) .await .change_context(Error::DownloadFossaCli) .describe("Broker relies on fossa-cli to perform analysis of your projects")?; @@ -346,21 +337,24 @@ async fn scan_git_reference( span_record!(scan_id, &job.scan_id); // Clone the reference into a temporary directory. - let cloned_location = job - .integration - .clone_reference(&job.reference) - .await - .change_context_lazy(|| Error::CloneReference(job.reference.clone()))?; + let cloned_location = retry_default( + format!("Clone '{}' at '{}'", job.integration, job.reference), + || async { job.integration.clone_reference(&job.reference).await }, + ) + .await + .change_context_lazy(|| Error::CloneReference(job.reference.clone()))?; // Record the CLI version for debugging purposes. let cli_version = cli.version().await.change_context(Error::RunFossaCli)?; span_record!(cli_version, display cli_version); // Run the scan. - let source_units = cli - .analyze(&job.scan_id, cloned_location.path()) - .await - .change_context(Error::RunFossaCli)?; + let source_units = retry_default( + format!("Analyze '{}' at '{}'", job.integration, job.reference), + || async { cli.analyze(&job.scan_id, cloned_location.path()).await }, + ) + .await + .change_context(Error::RunFossaCli)?; info!("Scanned '{}' at '{}'", job.integration, job.reference); Ok(UploadSourceUnits { @@ -384,9 +378,11 @@ async fn upload_scans( let meta = ProjectMetadata::new(&job.integration, &job.reference); info!("Uploading scan for project: '{meta}'"); - let locator = fossa::upload_scan(ctx.config.fossa_api(), &meta, &job.cli, job.source_units) - .await - .change_context(Error::TaskHandle)?; + let locator = retry_default(format!("Upload results for '{meta}'"), || async { + fossa::upload_scan(ctx.config.fossa_api(), &meta, &job.cli, &job.source_units).await + }) + .await + .change_context(Error::TaskHandle)?; debug!(scan_id = %job.scan_id, locator = %locator, "Uploaded scan"); info!("Uploaded scan for project '{meta}' as locator: '{locator}'"); @@ -394,3 +390,44 @@ async fn upload_scans( guard.commit().change_context(Error::TaskComplete)?; } } + +/// Retries the provided action with the provided label, using the default retry strategy. +/// +/// The default retry strategy: +/// - Retries using an exponential backoff delay, with a jitter to prevent thundering herds. +/// - The delay starts at 1 second. +/// - Retries a total of ten times. +/// +/// Each time the action fails, a warning is output using the label, +/// in the form `{label}: attempt failed, will retry. error: {error}`. +/// +/// If the overall process fails, the returned error has help text attached instructing users +/// to review the warnings in the logs for troubleshooting. +/// +/// If the process eventually succeeds, no error is returned, but warnings are still emitted. +async fn retry_default(label: S, action: A) -> Result +where + S: AsRef, + A: Fn() -> F, + F: Future>, + E: Context, +{ + let wrapped_action = || async { + match action().await { + Ok(result) => Ok(result), + Err(err) => { + warn!( + "{}: attempt failed, will retry. error: {err:#}", + label.as_ref() + ); + Err(err) + } + } + }; + + let strategy = ExponentialBackoff::from_millis(1000).map(jitter).take(10); + Retry::spawn(strategy, wrapped_action).await.help(indoc! {" + Each time this operation was attempted, it logged a warning; please review those + warnings in the logs for more details. + "}) +}