CI: Retry Artifact Upload Finalization (#8970)

This has been observed to be the most random error-prone part of the Rust build scripts.

This adds several retries to the patching of the artifact size (which finalizes the upload).
Additional diagnostics was added, so we observe if the retries are actually helping, so we can better understand the issue if this is not enough to fix it.

build/ci_utils/src/actions/artifacts/raw.rs

@@ -22,6 +22,7 @@ use tokio::io::AsyncReadExt;
 
 pub mod endpoints {
     use super::*;
+    use crate::io::retry;
     use reqwest::header::HeaderValue;
     use std::pin::Pin;
     use tokio::io::AsyncRead;
@@ -110,21 +111,21 @@ pub mod endpoints {
     #[context("Failed to finalize upload of the artifact `{}`.", artifact_name.as_ref())]
     pub async fn patch_artifact_size(
         json_client: &reqwest::Client,
-        artifact_url: Url,
+        artifact_url: &Url,
         artifact_name: impl AsRef<str>,
         size: usize,
     ) -> Result<PatchArtifactSizeResponse> {
         debug!("Patching the artifact `{}` size.", artifact_name.as_ref());
-        let artifact_url = artifact_url.clone();
-
-        let patch_request = json_client
-            .patch(artifact_url.clone())
-            .query(&[("artifactName", artifact_name.as_ref())]) // OsStr can be passed here, fails runtime
-            .json(&PatchArtifactSize { size });
-
-        // TODO retry
-        let response = patch_request.send().await?;
-        Ok(response.json().await?)
+        let artifact_name = artifact_name.as_ref();
+        retry(async move || {
+            let patch_request = json_client
+                .patch(artifact_url.clone())
+                .query(&[("artifactName", artifact_name)]) // OsStr can be passed here, fails runtime
+                .json(&PatchArtifactSize { size });
+            let response = patch_request.send().await?;
+            Ok(response.json().await?)
+        })
+        .await
     }
 
     pub async fn download_item(

build/ci_utils/src/actions/artifacts/run_session.rs

@@ -57,7 +57,7 @@ impl SessionClient {
     ) -> Result<PatchArtifactSizeResponse> {
         raw::endpoints::patch_artifact_size(
             &self.json_client,
-            self.artifact_url.clone(),
+            &self.artifact_url,
             artifact_name,
             total_size,
         )

build/ci_utils/src/actions/workflow.rs

@@ -108,3 +108,7 @@ impl Message {
 pub fn message(level: MessageLevel, text: impl AsRef<str>) {
     Message { level, text: text.as_ref().into() }.send()
 }
+
+pub fn warn(text: impl AsRef<str>) {
+    message(MessageLevel::Warning, text)
+}

build/ci_utils/src/io.rs

@@ -74,15 +74,33 @@ pub async fn download_and_extract(
     client::download_and_extract(&default(), url, output_dir).await
 }
 
-// pub async fn stream_to_file<E: Into<Box<dyn std::error::Error + Send + Sync>>>(
-//     stream: impl Stream<Item = std::result::Result<Bytes, E>> + Unpin,
-//     output_path: impl AsRef<Path>,
-// ) -> Result {
-//     let mut reader = tokio_util::io::StreamReader::new(stream.map_err(std::io::Error::other));
-//     let mut output = crate::fs::tokio::create(output_path).await?;
-//     tokio::io::copy(&mut reader, &mut output).await?;
-//     Ok(())
-// }
+/// Retry a given action until it succeeds or the maximum number of attempts is reached.
+pub async fn retry<Fn, Fut, Ret>(mut action: Fn) -> Result<Ret>
+where
+    Fn: FnMut() -> Fut,
+    Fut: Future<Output = Result<Ret>>, {
+    let growth_factor = 1.5;
+    let mut attempts = 5;
+    let mut delay = std::time::Duration::from_millis(500);
+    let max_delay = std::time::Duration::from_secs(10);
+
+    loop {
+        match action().await {
+            Ok(result) => return Ok(result),
+            Err(err) => {
+                let warning = format!("Failed to execute action: {err:?}");
+                crate::actions::workflow::warn(&warning);
+                warn!("{warning}");
+                if attempts == 0 {
+                    return Err(err);
+                }
+                attempts -= 1;
+                tokio::time::sleep(delay).await;
+                delay = delay.mul_f32(growth_factor).min(max_delay);
+            }
+        }
+    }
+}
 
 
 #[cfg(test)]