feat(upstreams): route by priority and package-name globs, not by alphabet

crates/silo-cli/src/main.rs

@@ -286,6 +286,17 @@ enum RepoCommand {
         basic_auth: Option<String>,
         #[arg(long, conflicts_with = "basic_auth")]
         bearer_token: Option<String>,
+        /// 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<String>,
         /// apk/pacman/deb: which architectures to sync. Repeat for several.
         #[arg(long = "arch")]
         arches: Vec<String>,
@@ -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<i32>,
+        /// Replaces the package-name globs this upstream answers for.
+        /// Repeat for several. See `add-upstream`.
+        #[arg(long = "package-pattern")]
+        package_patterns: Vec<String>,
+        /// 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 { "-" }

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<char> = pattern.chars().collect();
+    let name: Vec<char> = 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<String>) -> 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);

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

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![],

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 '{}';