From 3b4b02d2f52d290e664e6ed04dd447f5d2f4460e Mon Sep 17 00:00:00 2001 From: Alexander Birkner Date: Wed, 9 Sep 2026 16:47:47 +0200 Subject: [PATCH] feat(upstreams): route by priority and package-name globs, not by alphabet Two upstreams of one format were tried in name order, and every one of them was asked about every package. That is fine for interchangeable mirrors and wrong for anything else -- a vendor registry holding one scope next to a public registry holding the rest. It fails quietly rather than loudly. A vendor registry typically answers for public names too, by proxying or 302-redirecting to the public registry, and reqwest follows redirects: so the first upstream by name wins every package, the packages are cached through *its* entry, and the whole repo ends up attributed to it in `silo list`. A production instance with `fontawesome` and `npmjs` configured had 69,393 synced rows against npmjs's 7,098, and 118 of 124 stored packages -- @babel/core, @eslint/*, @nuxt/image -- recorded as coming from the vendor registry, which had in fact redirected every one of them to npmjs. Two columns, both defaulting to today's behaviour: - `priority`, tried highest first, ties broken by name, so the order is explicit instead of an accident of the alphabet and reordering does not mean renaming. - `package_patterns`, globs (`*`, `?`) the package name must match for an upstream to be consulted at all. Empty means no restriction, so nothing changes until an operator opts in. Repeatable, so one upstream can hold several scopes. The filter is applied everywhere the package name is known: the artifact miss path, npm's lazy per-name sync, and the index merge -- an upstream must not advertise a name it is not allowed to serve, or a client would ask for it and be turned away after the fact. `update_upstream` takes its settings as a struct rather than a tenth positional argument. Verified against a running server with a registry that 302s unknown names to npmjs, exactly as the vendor one does: without a pattern, `left-pad` and `ms` are stored as `upstream:vendor`; with `--package-pattern '@vendor/*'` they are `upstream:npmjs` and the vendor registry is not asked about them at all. Co-Authored-By: Claude Opus 5 (1M context) --- crates/silo-cli/src/main.rs | 54 +++++ crates/silo-core/src/pull_through.rs | 169 +++++++++++++++- crates/silo-core/src/repo.rs | 7 + crates/silo-core/src/upstream_sync.rs | 2 + .../migrations/0006_upstream_routing.sql | 19 ++ crates/silo-db/src/upstreams.rs | 163 ++++++++++++---- crates/silo-server/src/admin.rs | 78 +++++--- crates/silo-server/src/http.rs | 10 + crates/silo-server/tests/publish_flow.rs | 2 + .../tests/upstream_pull_through.rs | 184 ++++++++++++++++++ proto/silo/v1/admin.proto | 21 +- 11 files changed, 635 insertions(+), 74 deletions(-) create mode 100644 crates/silo-db/migrations/0006_upstream_routing.sql diff --git a/crates/silo-cli/src/main.rs b/crates/silo-cli/src/main.rs index bf841eb..84820bb 100644 --- a/crates/silo-cli/src/main.rs +++ b/crates/silo-cli/src/main.rs @@ -286,6 +286,17 @@ enum RepoCommand { basic_auth: Option, #[arg(long, conflicts_with = "basic_auth")] bearer_token: Option, + /// Tried highest first; ties fall back to the upstream name. + #[arg(long, default_value_t = 0)] + priority: i32, + /// Only consult this upstream for package names matching this + /// glob (`*`, `?`). Repeat for several — a name matching any one + /// of them is served. Without any, the upstream answers for every + /// name. Use it when two upstreams of one format are not + /// interchangeable mirrors, e.g. `--package-pattern '@acme/*'` + /// for a vendor registry sitting next to a public one. + #[arg(long = "package-pattern")] + package_patterns: Vec, /// apk/pacman/deb: which architectures to sync. Repeat for several. #[arg(long = "arch")] arches: Vec, @@ -335,6 +346,17 @@ enum RepoCommand { /// Same as `--clear-arches`, for the component list. #[arg(long, conflicts_with = "components")] clear_components: bool, + /// Tried highest first; ties fall back to the upstream name. + #[arg(long)] + priority: Option, + /// Replaces the package-name globs this upstream answers for. + /// Repeat for several. See `add-upstream`. + #[arg(long = "package-pattern")] + package_patterns: Vec, + /// Same as `--clear-arches`, for the package patterns — leaves + /// the upstream answering for every name again. + #[arg(long, conflicts_with = "package_patterns")] + clear_package_patterns: bool, }, /// Removes an upstream. Cached packages it produced keep serving, /// relabeled as local, unless `--prune` is given. Admin only. @@ -1344,6 +1366,8 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) -> arches, suite, components, + priority, + package_patterns, } => { if !cache && !no_cache { anyhow::bail!("one of --cache or --no-cache is required"); @@ -1359,6 +1383,8 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) -> let response = client .add_upstream(with_auth( AddUpstreamRequest { + priority, + package_patterns, repo, channel, name, @@ -1394,6 +1420,9 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) -> components, clear_arches, clear_components, + priority, + package_patterns, + clear_package_patterns, } => { let cache_mode = match (cache, no_cache) { (true, false) => Some(UpstreamCacheMode::Cache as i32), @@ -1418,6 +1447,9 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) -> let response = client .update_upstream(with_auth( UpdateUpstreamRequest { + priority, + package_patterns, + clear_package_patterns, repo, channel, name, @@ -1484,6 +1516,8 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) -> "base_url": u.base_url, "cache_mode": upstream_cache_mode_name(u.cache_mode), "cache_index_in_memory": u.cache_index_in_memory, + "priority": u.priority, + "package_patterns": u.package_patterns, "auth_configured": u.auth_configured, "status": u.status, "last_sync_at": u.last_sync_at, @@ -1493,11 +1527,16 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) -> .collect(); return print_json(&json!(upstreams)); } + // Listed in the order they are actually tried, with the two + // things that decide it — a repo whose upstreams are not + // interchangeable mirrors is unreadable without them. let mut table = Table::new(&[ "NAME", "FORMAT", "BASE_URL", "CACHE", + "PRIO", + "PACKAGES", "AUTH", "STATUS", "LAST_SYNC", @@ -1508,6 +1547,12 @@ async fn cmd_repo(config_path: &str, server: Option<&str>, cmd: RepoCommand) -> format_name(u.format), u.base_url.clone(), upstream_cache_mode_name(u.cache_mode).to_string(), + u.priority.to_string(), + if u.package_patterns.is_empty() { + "*".to_string() + } else { + u.package_patterns.join(",") + }, u.auth_configured.to_string(), u.status.clone(), timestamp(u.last_sync_at), @@ -1582,6 +1627,15 @@ fn print_upstream(upstream: Option<&UpstreamInfo>) { println!(" base_url: {}", u.base_url); println!(" cache_mode: {}", upstream_cache_mode_name(u.cache_mode)); println!(" index cache: {}", u.cache_index_in_memory); + println!(" priority: {}", u.priority); + println!( + " packages: {}", + if u.package_patterns.is_empty() { + "* (every name)".to_string() + } else { + u.package_patterns.join(", ") + } + ); println!( " auth: {}", if u.auth_configured { "configured" } else { "-" } diff --git a/crates/silo-core/src/pull_through.rs b/crates/silo-core/src/pull_through.rs index 41120c3..d66d880 100644 --- a/crates/silo-core/src/pull_through.rs +++ b/crates/silo-core/src/pull_through.rs @@ -141,13 +141,16 @@ fn fetch_action(cache_mode: CacheMode, upstream_requires_auth: bool) -> Action { } /// Lists which configured upstreams back a `(repo, channel, format)` -/// triple, in the order they should be tried. Multiple upstreams of the -/// same format may be configured (the user's "one or more" ask); they're -/// tried in name order — a deterministic, operator-controllable order -/// (rename to reorder) without a separate priority column to manage. -/// Callers must fall through to the next candidate on a confirmed miss -/// rather than stopping at the first one, or every upstream after the -/// first is unreachable in practice. +/// triple, in the order they should be tried: highest `priority` first, +/// ties broken by name so the order is always total and stable. Callers +/// must fall through to the next candidate on a confirmed miss rather +/// than stopping at the first one, or every upstream after the first is +/// unreachable in practice. +/// +/// This does not apply [`upstream_serves`] — the package name isn't +/// always known this early (an rpm request carries a filename, not a +/// name). Callers that do know it filter with `upstream_serves` as they +/// go. pub async fn select_upstreams( db: &Db, repo: &str, @@ -159,6 +162,64 @@ pub async fn select_upstreams( Ok(upstreams) } +/// Whether `upstream` is allowed to answer for `package_name`. +/// +/// An upstream with no patterns answers for everything, which is what +/// every upstream does until someone says otherwise. Patterns exist for +/// the case where two upstreams of one format are *not* interchangeable +/// mirrors: a vendor registry that holds one scope, next to a public one +/// that holds the rest. Without them the only thing deciding which +/// upstream serves a name is which one answers first — and an upstream +/// that proxies or redirects unknown names to the public registry +/// answers for everything, so it wins everything, and every package in +/// the repo ends up attributed to it. +pub fn upstream_serves(upstream: &UpstreamRow, package_name: &str) -> bool { + upstream.package_patterns.is_empty() + || upstream + .package_patterns + .iter() + .any(|pattern| glob_matches(pattern, package_name)) +} + +/// Matches `name` against a glob: `*` stands for any run of characters +/// (including none, and including `/`), `?` for exactly one, and +/// everything else is literal. The whole name must match, not a prefix — +/// `@acme/*` is a scope, not "anything starting with `@acme/`" that +/// could also match `@acme/x/../y`. +/// +/// `*` deliberately spans `/` so `@acme/*` covers the whole scope. npm +/// names have at most one separator and the other formats have none, so +/// there is no nesting for a stricter `*` to protect. +/// +/// Hand-rolled rather than pulled in: this is the entire feature, the +/// two metacharacters are the two an operator expects from a shell glob, +/// and backtracking on inputs this short is free. +fn glob_matches(pattern: &str, name: &str) -> bool { + let pattern: Vec = pattern.chars().collect(); + let name: Vec = name.chars().collect(); + // `star` remembers the last `*` and how much of `name` it had eaten, + // so a failed match resumes by letting that `*` swallow one more + // character instead of giving up. + let (mut p, mut n) = (0usize, 0usize); + let mut star: Option<(usize, usize)> = None; + while n < name.len() { + if p < pattern.len() && (pattern[p] == '?' || pattern[p] == name[n]) { + p += 1; + n += 1; + } else if p < pattern.len() && pattern[p] == '*' { + star = Some((p, n)); + p += 1; + } else if let Some((star_p, star_n)) = star { + p = star_p + 1; + n = star_n + 1; + star = Some((star_p, star_n + 1)); + } else { + return false; + } + } + pattern[p..].iter().all(|c| *c == '*') +} + /// Whether decrypting `upstream`'s stored credential (if any) would /// succeed — used to decide `ProxyUpstream` vs `RedirectToUpstream` /// without actually needing the plaintext at this point. @@ -280,6 +341,100 @@ mod tests { ); } + #[test] + fn a_glob_matches_the_whole_name_not_a_prefix() { + assert!(glob_matches("lodash", "lodash")); + assert!(!glob_matches("lodash", "lodash-es")); + assert!(!glob_matches("odash", "lodash")); + } + + #[test] + fn a_scope_glob_covers_the_scope_and_nothing_else() { + assert!(glob_matches( + "@fortawesome/*", + "@fortawesome/fontawesome-pro" + )); + assert!(glob_matches( + "@fortawesome/*", + "@fortawesome/vue-fontawesome" + )); + // The one this feature exists for: a public package must not fall + // inside a vendor registry's scope. + assert!(!glob_matches("@fortawesome/*", "@babel/core")); + assert!(!glob_matches("@fortawesome/*", "lodash")); + // `*` spans `/`, so a scope glob covers the whole scope, but the + // scope prefix itself still has to match exactly. + assert!(!glob_matches("@fortawesome/*", "@fortawesomeX/thing")); + } + + #[test] + fn a_star_matches_an_empty_run_and_a_question_mark_matches_exactly_one() { + assert!(glob_matches("*", "")); + assert!(glob_matches("*", "anything/at-all")); + assert!(glob_matches("@acme/*", "@acme/")); + assert!(glob_matches("nod?", "node")); + assert!(!glob_matches("nod?", "nod")); + assert!(!glob_matches("nod?", "nodejs")); + } + + #[test] + fn several_stars_and_trailing_literals_still_match() { + assert!(glob_matches("*-plugin-*", "eslint-plugin-vue")); + assert!(glob_matches("@*/core", "@babel/core")); + assert!(!glob_matches("@*/core", "@babel/parser")); + assert!(glob_matches("**", "anything")); + } + + #[test] + fn an_upstream_without_patterns_answers_for_everything() { + let mut upstream = test_upstream(vec![]); + assert!(upstream_serves(&upstream, "anything")); + upstream.package_patterns = vec!["@acme/*".into()]; + assert!(upstream_serves(&upstream, "@acme/widget")); + assert!(!upstream_serves(&upstream, "lodash")); + } + + #[test] + fn patterns_are_alternatives_so_one_upstream_can_hold_several_scopes() { + let upstream = test_upstream(vec![ + "@acme/*".into(), + "@vendor/*".into(), + "legacy-tool".into(), + ]); + assert!(upstream_serves(&upstream, "@acme/widget")); + assert!(upstream_serves(&upstream, "@vendor/thing")); + assert!(upstream_serves(&upstream, "legacy-tool")); + assert!(!upstream_serves(&upstream, "@other/thing")); + } + + fn test_upstream(package_patterns: Vec) -> UpstreamRow { + UpstreamRow { + id: silo_db::Uuid::nil(), + repo: "r".into(), + channel: "c".into(), + name: "n".into(), + format: "npm".into(), + base_url: "https://example.com".into(), + cache_mode: "cache".into(), + cache_index_in_memory: false, + priority: 0, + package_patterns, + arches: vec![], + suite: None, + components: vec![], + auth_kind: None, + auth_username: None, + auth_secret_ciphertext: None, + auth_secret_nonce: None, + status: "ok".into(), + last_sync_at: None, + last_sync_error: None, + last_success_at: None, + created_at: chrono::DateTime::from_timestamp(0, 0).unwrap(), + updated_at: chrono::DateTime::from_timestamp(0, 0).unwrap(), + } + } + #[test] fn cache_mode_parses_the_two_stored_strings_and_rejects_anything_else() { assert_eq!(CacheMode::parse("cache").unwrap(), CacheMode::Cache); diff --git a/crates/silo-core/src/repo.rs b/crates/silo-core/src/repo.rs index d49518e..d561761 100644 --- a/crates/silo-core/src/repo.rs +++ b/crates/silo-core/src/repo.rs @@ -507,6 +507,13 @@ async fn merge_upstream_records( // own group is regenerated; a synthetic upstream row needs // the same check applied explicitly, since it never goes // through that regeneration path itself. + // An upstream restricted to a set of package names must not + // advertise anything else, or a client would ask for a name + // this upstream is not supposed to serve and the miss path + // would have to turn it away after the fact. + if !crate::pull_through::upstream_serves(upstream, &row.name) { + continue; + } let belongs = match format { PackageFormat::Apk | PackageFormat::Pacman => { row.arch == index_group diff --git a/crates/silo-core/src/upstream_sync.rs b/crates/silo-core/src/upstream_sync.rs index 6e219a6..39f1e81 100644 --- a/crates/silo-core/src/upstream_sync.rs +++ b/crates/silo-core/src/upstream_sync.rs @@ -253,6 +253,8 @@ mod tests { base_url: "https://example.com/repo".into(), cache_mode: "cache".into(), cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], arches: vec![], suite: None, components: vec![], diff --git a/crates/silo-db/migrations/0006_upstream_routing.sql b/crates/silo-db/migrations/0006_upstream_routing.sql new file mode 100644 index 0000000..0d829ac --- /dev/null +++ b/crates/silo-db/migrations/0006_upstream_routing.sql @@ -0,0 +1,19 @@ +-- Which upstream answers for a given package, when a repo/channel has +-- more than one of the same format. +-- +-- Without these two columns the only ordering is the name, alphabetically, +-- and every upstream is asked about every package. That is wrong the +-- moment the upstreams are not interchangeable mirrors — a vendor registry +-- holding one scope alongside a public one, say. It also fails quietly +-- rather than loudly: an upstream that answers for a name it has no +-- business serving (proxying it, or redirecting to the public registry) +-- simply wins, and every package ends up attributed to it. + +-- Higher is tried first; ties fall back to the name, so an existing +-- deployment that has never set a priority keeps the order it has today. +ALTER TABLE upstreams ADD COLUMN priority INTEGER NOT NULL DEFAULT 0; + +-- Globs the package name must match for this upstream to be consulted at +-- all. Empty means "no restriction", which is what every existing row +-- gets, so adding this changes nothing until an operator opts in. +ALTER TABLE upstreams ADD COLUMN package_patterns TEXT[] NOT NULL DEFAULT '{}'; diff --git a/crates/silo-db/src/upstreams.rs b/crates/silo-db/src/upstreams.rs index 65361da..38f345d 100644 --- a/crates/silo-db/src/upstreams.rs +++ b/crates/silo-db/src/upstreams.rs @@ -22,6 +22,12 @@ pub struct UpstreamRow { pub base_url: String, pub cache_mode: String, pub cache_index_in_memory: bool, + /// Tried highest first. See `silo_core::pull_through::select_upstreams`. + pub priority: i32, + /// Globs the package name must match for this upstream to be + /// consulted. Empty means no restriction. See + /// `silo_core::pull_through::upstream_serves`. + pub package_patterns: Vec, pub arches: Vec, pub suite: Option, pub components: Vec, @@ -57,16 +63,34 @@ pub struct NewUpstream { pub base_url: String, pub cache_mode: String, pub cache_index_in_memory: bool, + pub priority: i32, + pub package_patterns: Vec, pub arches: Vec, pub suite: Option, pub components: Vec, pub auth: Option, } +/// The mutable half of an upstream: everything `update-upstream` can +/// change, as one value rather than a positional argument list nobody can +/// read at the call site. +#[derive(Debug, Clone)] +pub struct UpstreamSettings<'a> { + pub base_url: &'a str, + pub cache_mode: &'a str, + pub cache_index_in_memory: bool, + pub priority: i32, + pub package_patterns: &'a [String], + pub arches: &'a [String], + pub suite: Option<&'a str>, + pub components: &'a [String], +} + const COLUMNS: &str = "id, repo, channel, name, format, base_url, cache_mode, \ - cache_index_in_memory, arches, suite, components, auth_kind, \ - auth_username, auth_secret_ciphertext, auth_secret_nonce, status, \ - last_sync_at, last_sync_error, last_success_at, created_at, updated_at"; + cache_index_in_memory, priority, package_patterns, arches, suite, \ + components, auth_kind, auth_username, auth_secret_ciphertext, \ + auth_secret_nonce, status, last_sync_at, last_sync_error, \ + last_success_at, created_at, updated_at"; impl Db { /// Inserts a new upstream row. Fails on a `(repo, channel, name)` @@ -85,10 +109,10 @@ impl Db { }; Ok(sqlx::query_as(&format!( "INSERT INTO upstreams (repo, channel, name, format, base_url, cache_mode, \ - cache_index_in_memory, arches, suite, components, \ - auth_kind, auth_username, auth_secret_ciphertext, \ - auth_secret_nonce, status) \ - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,'ok') \ + cache_index_in_memory, priority, package_patterns, \ + arches, suite, components, auth_kind, auth_username, \ + auth_secret_ciphertext, auth_secret_nonce, status) \ + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,'ok') \ RETURNING {COLUMNS}" )) .bind(&new.repo) @@ -98,6 +122,8 @@ impl Db { .bind(&new.base_url) .bind(&new.cache_mode) .bind(new.cache_index_in_memory) + .bind(new.priority) + .bind(&new.package_patterns) .bind(&new.arches) .bind(&new.suite) .bind(&new.components) @@ -171,33 +197,30 @@ impl Db { /// stored credential untouched; clearing it is a distinct explicit /// action (`clear_upstream_auth`) so a caller can't accidentally wipe /// a working credential by omitting `auth` from an unrelated update. - #[allow(clippy::too_many_arguments)] pub async fn update_upstream( &self, id: Uuid, - base_url: &str, - cache_mode: &str, - cache_index_in_memory: bool, - arches: &[String], - suite: Option<&str>, - components: &[String], + settings: &UpstreamSettings<'_>, auth: Option<&SealedAuth>, ) -> anyhow::Result> { if let Some(auth) = auth { return Ok(sqlx::query_as(&format!( "UPDATE upstreams SET base_url = $2, cache_mode = $3, \ - cache_index_in_memory = $4, arches = $5, suite = $6, components = $7, \ - auth_kind = $8, auth_username = $9, auth_secret_ciphertext = $10, \ - auth_secret_nonce = $11, updated_at = now() \ + cache_index_in_memory = $4, priority = $5, package_patterns = $6, \ + arches = $7, suite = $8, components = $9, \ + auth_kind = $10, auth_username = $11, auth_secret_ciphertext = $12, \ + auth_secret_nonce = $13, updated_at = now() \ WHERE id = $1 RETURNING {COLUMNS}" )) .bind(id) - .bind(base_url) - .bind(cache_mode) - .bind(cache_index_in_memory) - .bind(arches) - .bind(suite) - .bind(components) + .bind(settings.base_url) + .bind(settings.cache_mode) + .bind(settings.cache_index_in_memory) + .bind(settings.priority) + .bind(settings.package_patterns) + .bind(settings.arches) + .bind(settings.suite) + .bind(settings.components) .bind(&auth.kind) .bind(&auth.username) .bind(&auth.ciphertext) @@ -207,25 +230,23 @@ impl Db { } Ok(sqlx::query_as(&format!( "UPDATE upstreams SET base_url = $2, cache_mode = $3, \ - cache_index_in_memory = $4, arches = $5, suite = $6, components = $7, \ - updated_at = now() \ + cache_index_in_memory = $4, priority = $5, package_patterns = $6, \ + arches = $7, suite = $8, components = $9, updated_at = now() \ WHERE id = $1 RETURNING {COLUMNS}" )) .bind(id) - .bind(base_url) - .bind(cache_mode) - .bind(cache_index_in_memory) - .bind(arches) - .bind(suite) - .bind(components) + .bind(settings.base_url) + .bind(settings.cache_mode) + .bind(settings.cache_index_in_memory) + .bind(settings.priority) + .bind(settings.package_patterns) + .bind(settings.arches) + .bind(settings.suite) + .bind(settings.components) .fetch_optional(self.pool()) .await?) } - /// Clears a stored credential without touching anything else — - /// `update_upstream` only ever *sets* one, so a caller that wants to - /// remove a credential entirely (rather than replace it) uses this - /// instead. pub async fn clear_upstream_auth(&self, id: Uuid) -> anyhow::Result<()> { sqlx::query( "UPDATE upstreams SET auth_kind = NULL, auth_username = NULL, \ @@ -558,7 +579,8 @@ pub async fn list_in_channel<'e, E: PgExecutor<'e>>( channel: &str, ) -> anyhow::Result> { Ok(sqlx::query_as(&format!( - "SELECT {COLUMNS} FROM upstreams WHERE repo = $1 AND channel = $2 ORDER BY name" + "SELECT {COLUMNS} FROM upstreams WHERE repo = $1 AND channel = $2 \ + ORDER BY priority DESC, name" )) .bind(repo) .bind(channel) @@ -633,6 +655,8 @@ mod tests { base_url: "https://example.com/repo".to_string(), cache_mode: "cache".to_string(), cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], arches: vec![], suite: None, components: vec![], @@ -681,6 +705,73 @@ mod tests { assert_eq!(listed[0].name, "epel"); } + /// Upstreams come back in the order they should be tried: highest + /// priority first, and the name only as a tie-break. Ordering by name + /// alone means the only way to reorder is to rename, and which + /// upstream serves a package is then an accident of the alphabet. + #[tokio::test] + async fn upstreams_are_listed_highest_priority_first_then_by_name() { + let Some(db) = db().await else { + eprintln!("skipping: set SILO_TEST_DATABASE_URL"); + return; + }; + let repo = unique("prio"); + for (name, priority) in [("aaa", 0), ("mmm", 10), ("zzz", 10), ("bbb", -5)] { + let mut new = new_upstream(&repo, "stable", name); + new.priority = priority; + db.create_upstream(&new).await.unwrap(); + } + + let listed = db.list_upstreams(&repo, "stable").await.unwrap(); + let order: Vec<&str> = listed.iter().map(|u| u.name.as_str()).collect(); + assert_eq!(order, vec!["mmm", "zzz", "aaa", "bbb"]); + } + + /// Both routing columns round-trip, and an upstream that has never + /// been given either behaves exactly as it did before they existed. + #[tokio::test] + async fn routing_settings_round_trip_and_default_to_unrestricted() { + let Some(db) = db().await else { + eprintln!("skipping: set SILO_TEST_DATABASE_URL"); + return; + }; + let repo = unique("routing"); + let plain = db + .create_upstream(&new_upstream(&repo, "stable", "plain")) + .await + .unwrap(); + assert_eq!(plain.priority, 0); + assert!(plain.package_patterns.is_empty()); + + let mut scoped = new_upstream(&repo, "stable", "scoped"); + scoped.priority = 7; + scoped.package_patterns = vec!["@acme/*".into(), "legacy-tool".into()]; + let scoped = db.create_upstream(&scoped).await.unwrap(); + assert_eq!(scoped.priority, 7); + assert_eq!(scoped.package_patterns, vec!["@acme/*", "legacy-tool"]); + + let updated = db + .update_upstream( + scoped.id, + &UpstreamSettings { + base_url: &scoped.base_url, + cache_mode: &scoped.cache_mode, + cache_index_in_memory: scoped.cache_index_in_memory, + priority: -3, + package_patterns: &[], + arches: &scoped.arches, + suite: scoped.suite.as_deref(), + components: &scoped.components, + }, + None, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(updated.priority, -3); + assert!(updated.package_patterns.is_empty()); + } + #[tokio::test] async fn duplicate_name_within_repo_channel_is_rejected() { let Some(db) = db().await else { diff --git a/crates/silo-server/src/admin.rs b/crates/silo-server/src/admin.rs index 6883bc7..e0f4ea4 100644 --- a/crates/silo-server/src/admin.rs +++ b/crates/silo-server/src/admin.rs @@ -1265,6 +1265,8 @@ impl AdminServiceImpl { )); } + validate_package_patterns(&req.package_patterns)?; + let opts = silo_pkg::UpstreamFetchOptions { arches: req.arches.clone(), suite: non_empty_str(&req.suite).map(str::to_string), @@ -1306,6 +1308,8 @@ impl AdminServiceImpl { base_url: req.base_url.clone(), cache_mode: cache_mode.to_string(), cache_index_in_memory: req.cache_index_in_memory, + priority: req.priority, + package_patterns: req.package_patterns.clone(), arches: req.arches.clone(), suite: opts.suite.clone(), components: req.components.clone(), @@ -1421,6 +1425,25 @@ impl AdminServiceImpl { } else { req.components.clone() }; + let priority = req.priority.unwrap_or(existing.priority); + let package_patterns = if req.clear_package_patterns { + Vec::new() + } else if req.package_patterns.is_empty() { + existing.package_patterns.clone() + } else { + req.package_patterns.clone() + }; + validate_package_patterns(&package_patterns)?; + let settings = silo_db::upstreams::UpstreamSettings { + base_url: &base_url, + cache_mode: &cache_mode, + cache_index_in_memory, + priority, + package_patterns: &package_patterns, + arches: &arches, + suite: suite.as_deref(), + components: &components, + }; let sealed = match &req.auth { None => None, @@ -1446,16 +1469,7 @@ impl AdminServiceImpl { None => { self.state .db - .update_upstream( - existing.id, - &base_url, - &cache_mode, - cache_index_in_memory, - &arches, - suite.as_deref(), - &components, - None, - ) + .update_upstream(existing.id, &settings, None) .await } Some(None) => { @@ -1470,31 +1484,13 @@ impl AdminServiceImpl { .map_err(|e| Status::internal(e.to_string()))?; self.state .db - .update_upstream( - existing.id, - &base_url, - &cache_mode, - cache_index_in_memory, - &arches, - suite.as_deref(), - &components, - None, - ) + .update_upstream(existing.id, &settings, None) .await } Some(Some(sealed)) => { self.state .db - .update_upstream( - existing.id, - &base_url, - &cache_mode, - cache_index_in_memory, - &arches, - suite.as_deref(), - &components, - Some(&sealed), - ) + .update_upstream(existing.id, &settings, Some(&sealed)) .await } } @@ -1736,6 +1732,26 @@ fn non_empty_str(s: &str) -> Option<&str> { /// would round-trip in plain text through `download_url` and, on a /// `no_cache` upstream, straight into a client-visible redirect /// `Location` header. +/// A pattern that can never match anything is always a mistake, and one +/// that silently routes every package to the wrong upstream — so it is +/// rejected at the point it is set rather than discovered later from the +/// `origin` column. +fn validate_package_patterns(patterns: &[String]) -> Result<(), Status> { + for pattern in patterns { + if pattern.trim().is_empty() { + return Err(Status::invalid_argument( + "a package pattern must not be empty", + )); + } + if pattern.trim() != pattern { + return Err(Status::invalid_argument(format!( + "package pattern `{pattern}` has leading or trailing whitespace", + ))); + } + } + Ok(()) +} + fn validate_base_url(url: &str) -> Result<(), Status> { if !url.starts_with("http://") && !url.starts_with("https://") { return Err(Status::invalid_argument( @@ -1794,6 +1810,8 @@ fn to_proto_upstream(row: &silo_db::upstreams::UpstreamRow) -> UpstreamInfo { base_url: row.base_url.clone(), cache_mode: cache_mode as i32, cache_index_in_memory: row.cache_index_in_memory, + priority: row.priority, + package_patterns: row.package_patterns.clone(), arches: row.arches.clone(), suite: row.suite.clone().unwrap_or_default(), components: row.components.clone(), diff --git a/crates/silo-server/src/http.rs b/crates/silo-server/src/http.rs index b9ae502..8809cc0 100644 --- a/crates/silo-server/src/http.rs +++ b/crates/silo-server/src/http.rs @@ -490,6 +490,13 @@ async fn pull_through_miss( let mut upstream_pkg = None; let mut matched_upstream = None; for upstream in upstreams { + // An upstream restricted to a set of package names is not asked + // about anything else, however eagerly it would answer. + if let Some(name) = package_name { + if !silo_core::pull_through::upstream_serves(upstream, name) { + continue; + } + } // When the in-memory cache is enabled for this upstream, read // through it — the same one the index merge uses, so the two // lookup paths never disagree about what's fresh. Otherwise this @@ -750,6 +757,9 @@ async fn lazy_sync_npm_upstream_packages( let mut fetched = Vec::new(); let mut saw_transient_error = false; for upstream in upstreams { + if !silo_core::pull_through::upstream_serves(upstream, name) { + continue; + } match silo_core::upstream_sync::sync_npm_package( &state.db, state.upstream_http.clone(), diff --git a/crates/silo-server/tests/publish_flow.rs b/crates/silo-server/tests/publish_flow.rs index 4034ca1..6313a78 100644 --- a/crates/silo-server/tests/publish_flow.rs +++ b/crates/silo-server/tests/publish_flow.rs @@ -1118,6 +1118,8 @@ async fn a_publish_holds_only_one_pooled_connection_at_a_time() { base_url: "https://registry.example".into(), cache_mode: "cache".into(), cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], arches: vec![], suite: None, components: vec![], diff --git a/crates/silo-server/tests/upstream_pull_through.rs b/crates/silo-server/tests/upstream_pull_through.rs index 8ebaaa9..f12991b 100644 --- a/crates/silo-server/tests/upstream_pull_through.rs +++ b/crates/silo-server/tests/upstream_pull_through.rs @@ -98,6 +98,8 @@ async fn add_upstream_validates_and_syncs_before_creating_the_row() { base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec!["x86_64".into()], suite: String::new(), @@ -144,6 +146,8 @@ async fn add_upstream_against_an_unreachable_url_creates_no_row() { base_url: "http://127.0.0.1:1".into(), // nothing listens here cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec!["x86_64".into()], suite: String::new(), @@ -206,6 +210,8 @@ async fn setup_cache_upstream( base_url: mock.uri(), cache_mode: cache_mode as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth, arches: vec!["x86_64".into()], suite: String::new(), @@ -624,6 +630,8 @@ async fn a_noarch_upstream_package_is_merged_into_every_concrete_architecture() base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec!["aarch64".into()], suite: String::new(), @@ -709,6 +717,8 @@ async fn cache_index_in_memory_does_not_change_pull_through_behavior() { base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: true, + priority: 0, + package_patterns: vec![], auth: None, arches: vec!["x86_64".into()], suite: String::new(), @@ -840,6 +850,8 @@ async fn npm_packument_miss_lazily_syncs_and_renders_from_the_upstream() { base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -973,6 +985,8 @@ async fn a_cache_mode_signed_rpm_upstream_is_not_advertised_until_actually_fetch base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -1093,6 +1107,8 @@ async fn a_second_upstream_is_still_reachable_when_an_earlier_one_by_name_does_n base_url: alpha_mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec!["x86_64".into()], suite: String::new(), @@ -1133,6 +1149,8 @@ async fn a_second_upstream_is_still_reachable_when_an_earlier_one_by_name_does_n base_url: zeta_mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec!["x86_64".into()], suite: String::new(), @@ -1199,6 +1217,8 @@ async fn npm_packument_miss_falls_through_to_a_second_upstream_by_name() { base_url: alpha_mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -1246,6 +1266,8 @@ async fn npm_packument_miss_falls_through_to_a_second_upstream_by_name() { base_url: zeta_mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -1352,6 +1374,8 @@ async fn a_transient_upstream_failure_surfaces_as_a_retryable_502_not_a_permanen base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -1413,6 +1437,8 @@ async fn an_upstream_confirmed_404_still_surfaces_as_an_ordinary_404() { base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -1498,6 +1524,8 @@ async fn a_tarball_request_with_no_prior_packument_fetch_still_lazily_syncs() { base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -1617,6 +1645,8 @@ async fn an_npm_packument_only_merges_its_own_names_upstream_versions() { base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -1720,6 +1750,8 @@ async fn a_tarball_filename_shared_by_two_scoped_packages_resolves_to_the_right_ base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -1822,6 +1854,8 @@ async fn a_tarball_first_request_resolves_in_one_pass() { base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -1935,6 +1969,8 @@ async fn an_npm_lazy_sync_drops_the_upstreams_cached_index() { base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: true, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -2029,6 +2065,8 @@ async fn pull_through_audit_entries_can_be_switched_off_without_silencing_real_p base_url: mock.uri(), cache_mode: UpstreamCacheMode::Cache as i32, cache_index_in_memory: false, + priority: 0, + package_patterns: vec![], auth: None, arches: vec![], suite: String::new(), @@ -2087,3 +2125,149 @@ async fn pull_through_audit_entries_can_be_switched_off_without_silencing_real_p ); } } + +/// An upstream restricted to a set of package-name globs is not consulted +/// for anything else, even when it would happily answer. +/// +/// This is the shape a vendor registry makes: it holds one scope, sits +/// next to a public registry that holds the rest, and — like a real one — +/// answers for public names too by proxying or redirecting to the public +/// registry. Nothing but a pattern stops it winning every name it is +/// asked about, and every package in the repo then gets attributed to it. +#[tokio::test] +async fn an_upstream_scoped_to_a_pattern_is_not_consulted_for_other_names() { + let url = require_db!(); + let harness = Harness::new(&url).await; + let repo = unique_repo("npmscoped"); + let admin = harness.admin_token().await; + let service = AdminServiceImpl { + state: harness.state.clone(), + }; + + // The vendor registry: it has its own scope, and answers for + // everything else as well. + let vendor = MockServer::start().await; + // The public registry. + let public = MockServer::start().await; + for mock in [&vendor, &public] { + Mock::given(method("GET")) + .and(path("/")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(serde_json::json!({ "db_name": "registry" })), + ) + .mount(mock) + .await; + } + + let packument = |host: &str, name: &str| { + serde_json::json!({ + "name": name, + "versions": { + "1.0.0": { + "name": name, + "version": "1.0.0", + "dist": { + "tarball": format!("{host}/{name}/-/{}-1.0.0.tgz", + name.rsplit('/').next().unwrap()), + "shasum": "abc123", + }, + }, + }, + }) + }; + for name in ["@vendor/widget", "lodash"] { + let file = format!("{}-1.0.0.tgz", name.rsplit('/').next().unwrap()); + Mock::given(method("GET")) + .and(path(format!("/{name}"))) + .respond_with(ResponseTemplate::new(200).set_body_json(packument(&vendor.uri(), name))) + .mount(&vendor) + .await; + Mock::given(method("GET")) + .and(path(format!("/{name}/-/{file}"))) + .respond_with(ResponseTemplate::new(200).set_body_bytes(build_test_npm(name, "1.0.0"))) + .mount(&vendor) + .await; + } + Mock::given(method("GET")) + .and(path("/lodash")) + .respond_with(ResponseTemplate::new(200).set_body_json(packument(&public.uri(), "lodash"))) + .mount(&public) + .await; + Mock::given(method("GET")) + .and(path("/lodash/-/lodash-1.0.0.tgz")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(build_test_npm("lodash", "1.0.0"))) + .mount(&public) + .await; + + // `vendor` sorts first by name *and* is given the higher priority, so + // nothing but the pattern keeps it away from `lodash`. + for (name, base_url, patterns, priority) in [ + ("vendor", vendor.uri(), vec!["@vendor/*".to_string()], 10), + ("public", public.uri(), vec![], 0), + ] { + service + .add_upstream(request( + AddUpstreamRequest { + repo: repo.clone(), + channel: "stable".into(), + name: name.into(), + format: silo_proto::v1::PackageFormat::Npm as i32, + base_url, + cache_mode: UpstreamCacheMode::Cache as i32, + cache_index_in_memory: false, + priority, + package_patterns: patterns, + auth: None, + arches: vec![], + suite: String::new(), + components: vec![], + }, + &admin.secret, + )) + .await + .expect("add_upstream"); + } + harness.db.set_repo_public(&repo, true).await.unwrap(); + + for (package, expected_upstream) in [("@vendor/widget", "vendor"), ("lodash", "public")] { + let file = format!("{}-1.0.0.tgz", package.rsplit('/').next().unwrap()); + let response = get( + &harness.state, + &format!("/{repo}/stable/npm/{package}/-/{file}"), + ) + .await; + assert_eq!( + response.status(), + axum::http::StatusCode::OK, + "GET {package}" + ); + + let origin = harness + .db + .list_packages(&repo, "stable", Some(silo_pkg::PackageFormat::Npm)) + .await + .unwrap() + .into_iter() + .find(|p| p.name == package) + .and_then(|p| p.origin_upstream_id) + .expect("the package was cached with an origin"); + let origin = harness.db.find_upstream(origin).await.unwrap().unwrap(); + assert_eq!( + origin.name, expected_upstream, + "{package} was served by the wrong upstream" + ); + } + + // The vendor registry was never even asked about the public name. + let asked_vendor_for_lodash = vendor + .received_requests() + .await + .unwrap_or_default() + .iter() + .any(|r| r.url.path() == "/lodash"); + assert!( + !asked_vendor_for_lodash, + "a pattern-scoped upstream must not be asked about a name outside its patterns" + ); +} diff --git a/proto/silo/v1/admin.proto b/proto/silo/v1/admin.proto index cd89c25..16dbe1a 100644 --- a/proto/silo/v1/admin.proto +++ b/proto/silo/v1/admin.proto @@ -56,7 +56,8 @@ service AdminService { // and creates nothing. rpc AddUpstream(AddUpstreamRequest) returns (AddUpstreamResponse); // Changes an existing upstream's settings in place (cache mode, index - // cache toggle, auth, arches/suite/components). Identity + // cache toggle, auth, priority, package patterns, arches/suite/ + // components). Identity // (repo/channel/name/format) can't be changed this way — add a new one // instead. rpc UpdateUpstream(UpdateUpstreamRequest) returns (UpdateUpstreamResponse); @@ -409,6 +410,11 @@ message UpstreamInfo { int64 last_sync_at = 14; string last_sync_error = 15; int64 last_success_at = 16; + // Tried highest first; ties broken by name. + int32 priority = 17; + // Globs the package name must match for this upstream to be consulted. + // Empty means no restriction. + repeated string package_patterns = 18; } message AddUpstreamRequest { @@ -427,6 +433,14 @@ message AddUpstreamRequest { string suite = 10; // deb only. repeated string components = 11; + // Tried highest first; ties broken by name. Defaults to 0, which keeps + // the plain by-name order for a channel where nobody sets one. + int32 priority = 12; + // Globs (`*`, `?`) the package name must match for this upstream to be + // consulted at all — `@acme/*` for a vendor registry that holds one + // scope next to a public registry that holds the rest. Empty means no + // restriction, which is every upstream until someone says otherwise. + repeated string package_patterns = 13; } message AddUpstreamResponse { @@ -454,6 +468,11 @@ message UpdateUpstreamRequest { repeated string components = 10; bool clear_arches = 11; bool clear_components = 12; + optional int32 priority = 13; + // Same "unset means unchanged" convention as arches/components, with + // `clear_package_patterns` to empty the list explicitly. + repeated string package_patterns = 14; + bool clear_package_patterns = 15; } message UpdateUpstreamResponse {