Imports: a name the source refuses becomes a failed item, and the scan goes on

crates/server/src/import/mod.rs

@@ -563,9 +563,10 @@ struct ItemSource {
     /// Raw: the content type the source lists.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     content_type: Option<String>,
-    /// Raw and apt: why the item fails without being copied (a path Forge
-    /// refuses; an index entry that names no `.deb` below the repository, a
-    /// component Forge does not publish).
+    /// Why the item fails without being copied (raw: a path Forge refuses;
+    /// apt: an index entry that names no `.deb` below the repository, a
+    /// component Forge does not publish; any format: a name the source
+    /// failed on while the scan read it).
     #[serde(default, skip_serializing_if = "Option::is_none")]
     error: Option<String>,
     /// apt: where the package is, below the import's URL (its index's
@@ -652,6 +653,19 @@ impl Found {
     fn missing(name: &str) -> Found {
         Found::component(name, String::new(), None, ItemSource::default())
     }
+
+    /// A name the source failed on while the scan read it: an item that
+    /// fails with `error`, so the import reports it and goes on.
+    fn failed(target: &Target, name: &str, error: String) -> Found {
+        let source = ItemSource {
+            error: Some(error),
+            ..ItemSource::default()
+        };
+        match target {
+            Target::Files { .. } => file_item(name, source),
+            _ => Found::component(name, String::new(), None, source),
+        }
+    }
 }
 
 /// One page of the scan: what it found, and the cursor of the next page
@@ -906,8 +920,20 @@ impl Run<'_> {
     async fn retrying<'f, T>(
         &self,
         what: &str,
-        mut read: impl FnMut() -> BoxFuture<'f, Result<T, SourceError>> + Send,
+        read: impl FnMut() -> BoxFuture<'f, Result<T, SourceError>> + Send,
     ) -> Result<T, End> {
+        self.retried(what, read)
+            .await
+            .map_err(|e| End::Failed(format!("{what}: {e}")))
+    }
+
+    /// A read of the source, retried with a backoff while its failure may
+    /// pass; the last failure once it does not.
+    async fn retried<'f, T>(
+        &self,
+        what: &str,
+        mut read: impl FnMut() -> BoxFuture<'f, Result<T, SourceError>> + Send,
+    ) -> Result<T, SourceError> {
         let mut attempt = 0;
         loop {
             attempt += 1;
@@ -917,7 +943,7 @@ impl Run<'_> {
                     tracing::warn!(import = self.import.id, error = %e, attempt, "{what} failed; retrying");
                     tokio::time::sleep(Duration::from_secs(1 << attempt)).await;
                 }
-                Err(e) => return Err(End::Failed(format!("{what}: {e}"))),
+                Err(e) => return Err(e),
             }
         }
     }
@@ -1270,7 +1296,11 @@ impl Run<'_> {
 
     /// The versions of the package ids `names`, or the tags of the images
     /// `names`, in their order. A name the source does not have is left
-    /// out, or, when the import `named` it, found as missing.
+    /// out, or, when the import `named` it, found as missing. A name the
+    /// source refuses or fails on (401, 403, 500, after the retries of a
+    /// failure that may pass) is found as an item that fails with the
+    /// reason, and the others are still read: one name does not fail the
+    /// scan. Only a source that cannot be reached at all fails the page.
     async fn expand(
         &self,
         ctx: &Ctx,
@@ -1281,7 +1311,7 @@ impl Run<'_> {
         // The reads are made up front: futures do nothing until polled.
         let reads: Vec<_> = names
             .iter()
-            .map(|name| self.expand_one(ctx, name, named, concurrency))
+            .map(|name| self.expand_read(ctx, name, named, concurrency))
             .collect();
         let pages: Vec<Vec<Found>> = futures_util::stream::iter(reads)
             .buffered(concurrency)
@@ -1290,6 +1320,41 @@ impl Run<'_> {
         Ok(pages.into_iter().flatten().collect())
     }
 
+    /// [`Run::expand_one`], retried; a failure on `name` that is not the
+    /// source being unreachable is found as an item that fails.
+    async fn expand_read(
+        &self,
+        ctx: &Ctx,
+        name: &str,
+        named: bool,
+        concurrency: usize,
+    ) -> Result<Vec<Found>, SourceError> {
+        self.read_name(&ctx.target, name, || {
+            self.expand_one(ctx, name, named, concurrency).boxed()
+        })
+        .await
+    }
+
+    /// `read` of the name `name`, retried; a failure that is not the source
+    /// being unreachable is found as an item that fails.
+    async fn read_name<'f>(
+        &self,
+        target: &Target,
+        name: &str,
+        read: impl FnMut() -> BoxFuture<'f, Result<Vec<Found>, SourceError>> + Send,
+    ) -> Result<Vec<Found>, SourceError> {
+        let what = format!("reading {name:?} at the source");
+        match self.retried(&what, read).await {
+            Ok(found) => Ok(found),
+            Err(e @ SourceError::Unreachable(_)) => Err(e),
+            Err(e) => {
+                tracing::warn!(import = self.import.id, name, error = %e, "import: the source fails on a name");
+                let error = format!("reading it at the source failed: {e}");
+                Ok(vec![Found::failed(target, name, error)])
+            }
+        }
+    }
+
     /// The versions of the package id `name`, the tags of the image `name`,
     /// or the file at the path `name`; for npm, the versions and dist-tags
     /// of the package `name`, or of every package of the scope `name`
@@ -1319,7 +1384,11 @@ impl Run<'_> {
                 }
                 let reads: Vec<_> = packages
                     .iter()
-                    .map(|package| npm_package(registry, package, true))
+                    .map(|package| {
+                        self.read_name(&ctx.target, package, || {
+                            npm_package(registry, package, true).boxed()
+                        })
+                    })
                     .collect();
                 let found: Vec<Vec<Found>> = futures_util::stream::iter(reads)
                     .buffered(concurrency)
@@ -1434,6 +1503,14 @@ impl Run<'_> {
         let copied = match &ctx.target {
             // A file's item has no version.
             Target::Files { source } => self.copy_file(ctx, source, &item).await,
+            // A name the source failed on while the scan read it fails as
+            // its item, and does not count towards the failures in a row
+            // that stop an import.
+            Target::Packages { .. } | Target::Npm { .. } | Target::Images { .. }
+                if item.version.is_empty() && source.error.is_some() =>
+            {
+                Ok(copied("failed", source.error.clone(), 0))
+            }
             Target::Apt {
                 source: from,
                 writer,

crates/server/tests/import.rs

@@ -21,6 +21,9 @@
 //!   cancelling and resuming; the preflight's refusals;
 //! * an import that names its packages or images, from a source without a
 //!   search or a catalog;
+//! * a package or image the source lists but refuses or fails on (a 403 for
+//!   a feed's versions, a 500 for a packument, a 401 for an image's tags) is
+//!   a failed item, and the scan goes on with the other names;
 //! * Docker: the catalog paged, a multi-platform index with its children, a
 //!   cosign signature tag, and a referrer from the referrers API or, without
 //!   it, through the referrers tag schema, digests kept;
@@ -124,6 +127,10 @@ struct SourceState {
     delay: Duration,
     /// npm registries, by repository.
     npm: HashMap<String, NpmSource>,
+    /// Paths (`/repository/...`) answered with this status instead, to the
+    /// import's own credentials: a resource they may not read, or one the
+    /// source fails on.
+    refused: HashMap<String, StatusCode>,
 }
 
 /// A version of an npm package at a mock registry.
@@ -343,6 +350,9 @@ async fn answer(state: &Mutex<SourceState>, base: &str, req: Request) -> Respons
             .into_response();
     }
     let path = req.uri().path().to_string();
+    if let Some(status) = state.lock().unwrap().refused.get(&path) {
+        return status.into_response();
+    }
     let query: HashMap<String, String> =
         url::form_urlencoded::parse(req.uri().query().unwrap_or("").as_bytes())
             .into_owned()
@@ -1904,6 +1914,160 @@ async fn named_packages_and_images_come_from_a_source_without_search_or_catalog(
     assert_eq!(done["progress"]["failed"], 1, "{done}");
 }
 
+/// The items of the import `id` by `name@version` (`name` alone without a
+/// version), and their state.
+async fn item_states(srv: &TestServer, repo: &str, id: i64) -> BTreeMap<String, String> {
+    items(srv, repo, id, None)
+        .await
+        .iter()
+        .map(|item| {
+            let name = item["name"].as_str().unwrap();
+            let version = item["version"].as_str().unwrap_or_default();
+            let key = if version.is_empty() {
+                name.to_string()
+            } else {
+                format!("{name}@{version}")
+            };
+            (key, item["state"].as_str().unwrap().to_string())
+        })
+        .collect()
+}
+
+/// The detail of the failed item `name`.
+async fn failure_of(srv: &TestServer, repo: &str, id: i64, name: &str) -> String {
+    let failed = items(srv, repo, id, Some("failed")).await;
+    let item = failed
+        .iter()
+        .find(|item| item["name"] == name)
+        .unwrap_or_else(|| panic!("{name} is no failed item: {failed:?}"));
+    item["detail"].as_str().unwrap_or_default().to_string()
+}
+
+#[tokio::test]
+async fn a_package_the_feed_refuses_is_a_failed_item_and_the_search_goes_on() {
+    let srv = forge().await;
+    let source = Source::start().await;
+    let feed = "nuget-refusing";
+    for id in ["Acme.A", "Acme.B", "Acme.C"] {
+        source.nuget(
+            feed,
+            id,
+            "1.0.0",
+            make_nupkg(id, "1.0.0"),
+            "2021-01-01T00:00:00.000+00:00",
+        );
+    }
+    // The search lists Acme.B, but its versions are forbidden to the
+    // import's user.
+    source.with(|s| {
+        s.refused.insert(
+            format!("/repository/{feed}/flat/acme.b/index.json"),
+            StatusCode::FORBIDDEN,
+        );
+    });
+
+    let repo = repo_name("nuget-refused");
+    create_repo(&srv, json!({"name": repo, "kind": "hosted"})).await;
+    let id = start(&srv, &repo, from(source.feed_url(feed))).await;
+    let done = run(&srv, &repo, id).await;
+    assert_eq!(done["state"], "completed", "{done}");
+    assert_eq!(done["progress"]["imported"], 2, "{done}");
+    assert_eq!(done["progress"]["failed"], 1, "{done}");
+    let states = item_states(&srv, &repo, id).await;
+    assert_eq!(states["Acme.A@1.0.0"], "imported", "{states:?}");
+    assert_eq!(states["Acme.B"], "failed", "{states:?}");
+    assert_eq!(states["Acme.C@1.0.0"], "imported", "{states:?}");
+    let detail = failure_of(&srv, &repo, id, "Acme.B").await;
+    assert!(detail.contains("403"), "{detail}");
+    assert!(nupkg(&srv, &repo, "Acme.A", "1.0.0").await.is_some());
+    assert!(nupkg(&srv, &repo, "Acme.C", "1.0.0").await.is_some());
+}
+
+#[tokio::test]
+async fn a_package_the_npm_registry_fails_on_is_a_failed_item_and_the_names_go_on() {
+    let srv = forge().await;
+    let source = Source::start().await;
+    let reg = "npm-failing";
+    for name in ["alpha", "beta", "gamma"] {
+        source.npm_version(
+            reg,
+            name,
+            "1.0.0",
+            tgz(name, "1.0.0", name),
+            "2022-02-02T00:00:00.000Z",
+        );
+        source.npm_tag(reg, name, "latest", "1.0.0");
+    }
+    // The registry fails on beta's packument, every time.
+    source.with(|s| {
+        s.refused.insert(
+            format!("/repository/{reg}/beta"),
+            StatusCode::INTERNAL_SERVER_ERROR,
+        );
+    });
+
+    let repo = repo_name("npm-failing");
+    create_repo(
+        &srv,
+        json!({"name": repo, "kind": "hosted", "format": "npm"}),
+    )
+    .await;
+    let mut body = from(source.npm_url(reg));
+    body["names"] = json!(["alpha", "beta", "gamma"]);
+    let id = start(&srv, &repo, body).await;
+    let done = run(&srv, &repo, id).await;
+    assert_eq!(done["state"], "completed", "{done}");
+    assert_eq!(done["progress"]["failed"], 1, "{done}");
+    let states = item_states(&srv, &repo, id).await;
+    assert_eq!(states["alpha@1.0.0"], "imported", "{states:?}");
+    assert_eq!(states["beta"], "failed", "{states:?}");
+    assert_eq!(states["gamma@1.0.0"], "imported", "{states:?}");
+    assert!(!states.keys().any(|k| k.starts_with("beta@")), "{states:?}");
+    let detail = failure_of(&srv, &repo, id, "beta").await;
+    assert!(detail.contains("500"), "{detail}");
+    assert!(npm_tarball(&srv, &repo, "gamma", "1.0.0").await.is_some());
+}
+
+#[tokio::test]
+async fn an_image_whose_tags_are_refused_is_a_failed_item_and_the_catalog_goes_on() {
+    let srv = forge().await;
+    let source = Source::start().await;
+    let reg = "docker-rules";
+    // More images than a catalog page holds; team-b/app is listed, but its
+    // tags are not the import's user's to read.
+    for name in ["team-a/app", "team-b/app", "tools/cli"] {
+        let cfg = config("amd64");
+        let layer = format!("layer of {name}").into_bytes();
+        source.blob(reg, name, &cfg);
+        source.blob(reg, name, &layer);
+        source.tag(reg, name, "1.0", OCI_MANIFEST, &image(&cfg, &layer));
+    }
+    source.with(|s| {
+        s.refused.insert(
+            format!("/repository/{reg}/v2/team-b/app/tags/list"),
+            StatusCode::UNAUTHORIZED,
+        );
+    });
+
+    let repo = repo_name("docker-refused");
+    create_repo(
+        &srv,
+        json!({"name": repo, "kind": "hosted", "format": "docker"}),
+    )
+    .await;
+    let id = start(&srv, &repo, from(source.registry_url(reg))).await;
+    let done = run(&srv, &repo, id).await;
+    assert_eq!(done["state"], "completed", "{done}");
+    assert_eq!(done["progress"]["imported"], 2, "{done}");
+    assert_eq!(done["progress"]["failed"], 1, "{done}");
+    let states = item_states(&srv, &repo, id).await;
+    assert_eq!(states["team-a/app@1.0"], "imported", "{states:?}");
+    assert_eq!(states["team-b/app"], "failed", "{states:?}");
+    assert_eq!(states["tools/cli@1.0"], "imported", "{states:?}");
+    let detail = failure_of(&srv, &repo, id, "team-b/app").await;
+    assert!(detail.contains("refused"), "{detail}");
+}
+
 // ── Docker ──────────────────────────────────────────────────────────────────
 
 fn config(arch: &str) -> Vec<u8> {

docs/changes/2.2.0/752.fixed.md

@@ -0,0 +1,8 @@
+An import's scan failed as a whole, and every resume failed again at the
+same place, when the source refused or failed on one package or image it
+listed: a registry's `tags/list` answering 401 for an image the import's user
+may not pull, or a NuGet feed or npm registry answering 403 or 500 for one
+package. That name is now an item that fails with the source's answer, after
+the retries of a failure that may pass, and the scan goes on with the next
+name. Only a failure of the listing itself (a catalog, a search, `/-/all`), or
+a source that cannot be reached at all, fails the import.