diff --git a/crates/server/src/import/mod.rs b/crates/server/src/import/mod.rs index d59da8c8..12a5f037 100644 --- a/crates/server/src/import/mod.rs +++ b/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, - /// 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, /// 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> + Send, + read: impl FnMut() -> BoxFuture<'f, Result> + Send, ) -> Result { + 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> + Send, + ) -> Result { 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> = 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, 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, SourceError>> + Send, + ) -> Result, 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> = 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, diff --git a/crates/server/tests/import.rs b/crates/server/tests/import.rs index fd906a7d..8a650439 100644 --- a/crates/server/tests/import.rs +++ b/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, + /// 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, } /// A version of an npm package at a mock registry. @@ -343,6 +350,9 @@ async fn answer(state: &Mutex, 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 = 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 { + 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 { diff --git a/docs/changes/2.2.0/752.fixed.md b/docs/changes/2.2.0/752.fixed.md new file mode 100644 index 00000000..b21770fd --- /dev/null +++ b/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.