fix(tools): HTTP timeouts + bound SBOM-browser concurrency

tools/harbor-sbom-browser/src/handlers/artifact_tree.rs

@@ -2,22 +2,30 @@ use crate::structs::*;
 use axum::extract::State;
 use axum::http::StatusCode;
 use axum::response::{Html, IntoResponse, Response};
+use futures::{StreamExt, TryStreamExt};
 use lazy_static::lazy_static;
 use regex::Regex;
 use snafu::{OptionExt, ResultExt, Snafu};
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::{Arc, Mutex};
+use std::time::Duration;
 use strum::{EnumDiscriminants, IntoStaticStr};
 use tracing::error;
 use urlencoding::encode;
 
 type ArtifactTree = BTreeMap<String, BTreeMap<String, BTreeSet<TagInfo>>>;
 
+/// Cap on how many repositories are fetched from Harbor at once, so a cache
+/// miss can't fan out one request-chain per `sdp/` repository simultaneously.
+const MAX_CONCURRENT_REPO_REQUESTS: usize = 8;
+
 #[derive(Snafu, Debug, EnumDiscriminants)]
 #[strum_discriminants(derive(IntoStaticStr))]
 #[snafu(visibility(pub))]
 #[allow(clippy::enum_variant_names)]
 pub enum ArtifactTreeError {
+    #[snafu(display("cannot build the HTTP client"))]
+    BuildHttpClient { source: reqwest::Error },
     #[snafu(display("cannot get repositories"))]
     GetRepositories { source: reqwest::Error },
     #[snafu(display("cannot parse repositories"))]
@@ -179,7 +187,13 @@ async fn build_artifact_tree() -> Result<ArtifactTree, ArtifactTreeError> {
     let base_url = format!("https://{}/api/v2.0", registry_hostname);
     let url = format!("{}/repositories?page_size={}&q=name=~sdp/", base_url, 100);
 
-    let response = reqwest::get(&url).await.context(GetRepositoriesSnafu)?;
+    // Shared client with a timeout so a hung registry can't block a request.
+    let client = reqwest::Client::builder()
+        .timeout(Duration::from_secs(30))
+        .build()
+        .context(BuildHttpClientSnafu)?;
+
+    let response = client.get(&url).send().await.context(GetRepositoriesSnafu)?;
     let repositories: Vec<Repository> = response.json().await.context(ParseRepositoriesSnafu)?;
     let artifact_tree = Arc::new(Mutex::new(ArtifactTree::new()));
 
@@ -190,20 +204,26 @@ async fn build_artifact_tree() -> Result<ArtifactTree, ArtifactTreeError> {
             .split_once('/')
             .context(UnexpectedRepositoryNameSnafu)?;
         requests.push(process_artifacts(
+            &client,
             &base_url,
             project_name,
             repository_name,
             artifact_tree.clone(),
         ));
     }
-    futures::future::try_join_all(requests).await?;
+    // Bound concurrency instead of firing every repository's request-chain at once.
+    futures::stream::iter(requests)
+        .buffer_unordered(MAX_CONCURRENT_REPO_REQUESTS)
+        .try_collect::<Vec<()>>()
+        .await?;
     Ok(Arc::try_unwrap(artifact_tree)
         .unwrap()
         .into_inner()
         .unwrap())
 }
 
 pub async fn process_artifacts(
+    client: &reqwest::Client,
     base_url: &str,
     project_name: &str,
     repository_name: &str,
@@ -213,19 +233,21 @@ pub async fn process_artifacts(
     let mut page = 1;
     let page_size = 20;
     loop {
-        let artifacts_page: Vec<Artifact> = reqwest::get(format!(
-            "{}/projects/{}/repositories/{}/artifacts?page_size={}&page={}",
-            base_url,
-            encode(project_name),
-            encode(repository_name),
-            page_size,
-            page
-        ))
-        .await
-        .context(GetArtifactsSnafu)?
-        .json()
-        .await
-        .context(ParseArtifactsSnafu)?;
+        let artifacts_page: Vec<Artifact> = client
+            .get(format!(
+                "{}/projects/{}/repositories/{}/artifacts?page_size={}&page={}",
+                base_url,
+                encode(project_name),
+                encode(repository_name),
+                page_size,
+                page
+            ))
+            .send()
+            .await
+            .context(GetArtifactsSnafu)?
+            .json()
+            .await
+            .context(ParseArtifactsSnafu)?;
 
         let number_of_returned_artifacts = artifacts_page.len();
         artifacts.extend(artifacts_page);

tools/monitor-oci-artifacts/src/main.rs

@@ -3,6 +3,7 @@ use serde::Deserialize;
 use std::{
     fmt::Formatter,
     process::{exit, Command, Stdio},
+    time::Duration,
 };
 use urlencoding::encode;
 
@@ -58,13 +59,18 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let mut page = 1;
     let attestation_tag_regex = Regex::new(r"^sha256-[0-9a-f]{64}.att$").unwrap();
 
+    // Timeout so a hung registry can't block the job indefinitely.
+    let client = reqwest::Client::builder()
+        .timeout(Duration::from_secs(30))
+        .build()?;
+
     loop {
         let url = format!(
             "{}/repositories?page_size={}&page={}",
             base_url, page_size, page
         );
 
-        let response = reqwest::get(&url).await?;
+        let response = client.get(&url).send().await?;
         let repositories: Vec<Repository> = response.json().await?;
 
         for repository in &repositories {
@@ -80,17 +86,19 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
             loop {
                 // Loop over pages to get all artifacts
-                let artifacts_page: Vec<Artifact> = reqwest::get(format!(
-                    "{}/projects/{}/repositories/{}/artifacts?page_size={}&page={}",
-                    base_url,
-                    encode(project_name),
-                    encode(repository_name),
-                    page_size,
-                    page
-                ))
-                .await?
-                .json()
-                .await?;
+                let artifacts_page: Vec<Artifact> = client
+                    .get(format!(
+                        "{}/projects/{}/repositories/{}/artifacts?page_size={}&page={}",
+                        base_url,
+                        encode(project_name),
+                        encode(repository_name),
+                        page_size,
+                        page
+                    ))
+                    .send()
+                    .await?
+                    .json()
+                    .await?;
 
                 let number_of_returned_artifacts = artifacts_page.len();
                 artifacts.extend(artifacts_page);

tools/nexus-garbage-collector/src/main.rs

@@ -1,4 +1,4 @@
-use std::{env, process::exit};
+use std::{env, process::exit, time::Duration};
 
 use log::{debug, error, info};
 use regex::Regex;
@@ -35,7 +35,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let attestation_tag_regex = Regex::new(r"/manifests/sha256-([0-9a-f]{64}).att$").unwrap();
     let signature_tag_regex = Regex::new(r"/manifests/sha256-([0-9a-f]{64}).sig$").unwrap();
     let mut continuation_token: Option<String> = None;
-    let client = reqwest::Client::new();
+    // Timeout so a hung/slow Nexus can't block the job indefinitely.
+    let client = reqwest::Client::builder()
+        .timeout(Duration::from_secs(30))
+        .build()?;
 
     let mut signatures = Vec::<(String, String)>::new();
     let mut attestations = Vec::<(String, String)>::new();