From fe9f4e6cff2708acb1561663085bb79d8d854fe7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 30 Oct 2026 06:27:01 +0000 Subject: [PATCH] Group search: merge local members on ids, load the page's hits alone A group asked each local member (hosted, or a proxy searching its cache, as browse does) for its first 1,000 hits: every version of 1,000 ids and their casing, on every page and search, of which the merge used the ids. A member's ids past those were unreachable through the group, and its total stopped there. Local members now answer with the ids of all their hits (RepoReadOps::search_ids: the hosted and proxy backends return the id list their search computes anyway). The group merges, counts and pages the ids, then loads the versions and casing of its page alone, each hit through the member that won it (RepoReadOps::search_hits). The defaults of both page through search_unenriched, so any implementation stays correct. Members are searched four at a time instead of one after another. Since every id is now evaluated, the guard walk takes two shortcuts: an unfiltered graph memoizes one set of visible leaves for all ids, and an id no guard is active for is served by its visible leaves without a guarded walk. Fixes #536 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012e6DJXRhLYxFjzrWiwKFKc --- crates/core/src/repo/group.rs | 284 ++++++++++++++++++++------ crates/core/src/repo/hosted.rs | 82 ++++++-- crates/core/src/repo/mod.rs | 108 +++++++++- crates/core/src/repo/proxy.rs | 117 ++++++++--- crates/core/tests/group_read.rs | 346 +++++++++++++++++++++++++++++++- crates/server/tests/browse.rs | 111 ++++++++++ docs/changes/2.2.0/536.fixed.md | 8 + 7 files changed, 944 insertions(+), 112 deletions(-) create mode 100644 docs/changes/2.2.0/536.fixed.md diff --git a/crates/core/src/repo/group.rs b/crates/core/src/repo/group.rs index 0bfef05e..ac2d66e1 100644 --- a/crates/core/src/repo/group.rs +++ b/crates/core/src/repo/group.rs @@ -90,19 +90,23 @@ //! while walking and a repeated group is expanded only when reached at a //! strictly shallower depth or under other active guards. A repeated leaf is queried once at its first //! package-visible sighting. Search also queries each unique leaf once (a -//! proxy searching its upstream again for more hits, see below) and uses -//! the same package-specific walk to select the highest-priority winner. -//! Members answer search **unenriched** ([`RepoReadOps::search_unenriched`]); -//! the group merges, counts and pages their hits, then enriches only the final +//! local member again for the hits of the final page, a proxy searching its +//! upstream again for more hits, see below) and uses the same +//! package-specific walk to select the highest-priority winner. Members +//! answer search **unenriched** ([`RepoReadOps::search_unenriched`]); the +//! group merges, counts and pages their hits, then enriches only the final //! page through the member that won each hit, so per-hit manifest reads are //! bounded by the client's `take`, not by every member's result. //! //! ## Search //! A group pages the merge of its members' hits, so it asks each member for -//! its first hits, not for the client's page. A local member (hosted, or a -//! proxy searching its cache: an offline one, or for a search without the -//! upstream, like browse's) is asked for its first -//! [`SearchParams::MAX_TAKE`]. A proxy searching its upstream is asked for +//! its first hits, not for the client's page, a few members at a time. A +//! local member (hosted, or a proxy searching its cache: an offline one, or +//! for a search without the upstream, like browse's) is asked for the ids +//! of all its hits ([`RepoReadOps::search_ids`]), without their versions or +//! casing: the group merges, counts and pages the ids, then loads the hits +//! of its page alone, each through the member that won it +//! ([`RepoReadOps::search_hits`]). A proxy searching its upstream is asked for //! its first `skip + take`, rounded up to a power of two //! (`upstream_window`). While the hits returned do not settle the page (the //! proxy or the group's guards left some out), the searches that returned @@ -113,7 +117,8 @@ //! for `MAX_TAKE` at once. //! //! The merge serves each id once, from the highest-priority member whose -//! hits include it. The local members' ids come first, ordered by id; then +//! hits include it. The local members' ids come first, ordered by id (all +//! of them, so the group pages through every id its members hold); then //! the upstream searches' ids, one hit of each search in turn, in the //! upstreams' (relevance) order. An id keeps the place its first hit gives //! it, so more hits only add ids after the settled ones: successive pages @@ -310,6 +315,15 @@ impl GroupGraph { /// root lets through. A group whose guard is active for the id /// ([`Self::guard_active`]) lets through only its owners. fn serving_leaves(&self, package_id: &str, id: &IdGuards<'_>) -> Vec { + // Most ids meet no active guard: then the walk admits every visible + // leaf, and the memoized visible leaves answer without one. + if self + .guards + .keys() + .all(|group_id| self.guard_active(*group_id, id).is_none()) + { + return self.visible_leaves(package_id).ordered.clone(); + } let key = crate::format::lower_invariant(package_id); let mut path = HashSet::from([self.root_id]); let mut best_group_depth = HashMap::from([((self.root_id, Vec::new()), 0)]); @@ -332,7 +346,13 @@ impl GroupGraph { /// Visible leaf ids for `package_id`, in the same package-specific stable /// pre-order as the former path-expanding traversal. fn visible_leaves(&self, package_id: &str) -> Arc { - let key = crate::format::lower_invariant(package_id); + // Without package filters every id sees the same leaves: one entry + // serves them all (a group search asks for every id of its members). + let key = if self.unfiltered { + String::new() + } else { + crate::format::lower_invariant(package_id) + }; if let Some(cached) = self .visible_leaf_cache .lock() @@ -857,17 +877,21 @@ impl GroupRepo { /// production backend's single corpus scan rather than repeatedly /// re-running the whole group search for successive filtered pages. /// - /// A proxy that searches its upstream is asked for the leading window of - /// its hits the page needs ([`upstream_window`]), and for a larger one - /// while the hits it returned do not settle the page; see the module docs - /// for the order and the count. Members answer unenriched; only the - /// merged, paged result is enriched ([`Self::enrich_page`]). + /// A local member is asked for its ids, and for the hits of those it won + /// on the final page ([`Self::page_hits`]). A proxy that searches its + /// upstream is asked for the leading window of its hits the page needs + /// ([`upstream_window`]), and for a larger one while the hits it returned + /// do not settle the page; see the module docs for the order and the + /// count. Members answer unenriched; only the merged, paged result is + /// enriched ([`Self::enrich_page`]). async fn search_filtered( &self, repo: &str, params: &SearchParams, filter: &PackageFilter, ) -> Result { + use futures_util::StreamExt; + let graph = self.group_graph(repo).await?; let wanted = params.skip.saturating_add(params.effective_take()); @@ -875,30 +899,34 @@ impl GroupRepo { // `graph.leaves[i]`. let mut member_ops: Vec> = Vec::with_capacity(graph.leaves.len()); let mut filters: Vec = Vec::with_capacity(graph.leaves.len()); + for leaf in &graph.leaves { + member_ops.push(self.member_ops(leaf)?); + filters.push(graph.filter_for_leaf(leaf.id).intersect(filter.clone())); + } + // The members' reads are independent: a few at a time. + let reads: Vec<_> = graph + .leaves + .iter() + .zip(member_ops.iter().zip(&filters)) + .map(|(leaf, (ops, leaf_filter))| first_search(ops, leaf, params, leaf_filter, wanted)) + .collect(); + let outcomes: Vec> = futures_util::stream::iter(reads) + .buffered(MEMBER_SEARCHES) + .collect() + .await; let mut searches: Vec = Vec::with_capacity(graph.leaves.len()); let mut failures = MemberFailures::default(); - for leaf in &graph.leaves { - let ops = self.member_ops(leaf)?; - let leaf_filter = graph.filter_for_leaf(leaf.id).intersect(filter.clone()); - // An offline proxy searches its cache, like a hosted member. - let upstream = params.include_upstream && leaf.kind() == RepoKind::Proxy && leaf.online; - // A filter may hide most of the upstream's hits, and the proxy - // then counts only those it returns: it is asked for them all. - let window = (upstream && matches!(leaf_filter, PackageFilter::AllowAll)) - .then(|| upstream_window(wanted)); - let search = match search_member(&ops, leaf, params, window, &leaf_filter).await { - Ok(result) => { + for (leaf, outcome) in graph.leaves.iter().zip(outcomes) { + searches.push(match outcome { + Ok(search) => { failures.ok(); - MemberSearch::new(upstream, window, result) + search } Err(e) => { failures.failed(&leaf.name, e); - MemberSearch::failed(upstream) + MemberSearch::failed(searches_upstream(params, leaf)) } - }; - member_ops.push(ops); - filters.push(leaf_filter); - searches.push(search); + }); } failures.into_result()?; @@ -913,8 +941,8 @@ impl GroupRepo { if !graph.may_yield(leaf.id) { continue; } - for hit in &search.hits { - let lower = self.lower_id(&hit.original_id); + for id in &search.ids { + let lower = self.lower_id(id); if looked_up.insert(lower.clone()) { lookup.push(lower); } @@ -931,10 +959,10 @@ impl GroupRepo { let shortest = searches .iter() .filter(|search| search.growable()) - .map(|search| search.hits.len()) + .map(|search| search.ids.len()) .min(); for (member, search) in searches.iter_mut().enumerate() { - if !search.growable() || Some(search.hits.len()) != shortest { + if !search.growable() || Some(search.ids.len()) != shortest { continue; } let Some(window) = search.window else { @@ -958,7 +986,7 @@ impl GroupRepo { .await { Ok(result) if search.extended_by(&result) => { - *search = MemberSearch::new(search.upstream, Some(window), result) + *search = MemberSearch::upstream(Some(window), result) } // A proxy whose upstream search failed answers from its // cache instead, with other hits. @@ -975,20 +1003,82 @@ impl GroupRepo { }; let total_hits = merged.order.len().saturating_add(merged.unseen); - let page: Vec = merged + let page: Vec<(usize, usize)> = merged .order .into_iter() .skip(params.skip) .take(params.effective_take()) - .map(|(member, position)| RankedHit { - member, - hit: std::mem::take(&mut searches[member].hits[position]), - }) .collect(); + let page = self + .page_hits(&graph, &member_ops, &mut searches, params, page) + .await?; let hits = self.enrich_page(&graph, &member_ops, page).await; Ok(SearchResult { total_hits, hits }) } + /// The hits of the final page, given as the `(member, position)` of each + /// id in `searches`: an upstream search's hit as it returned it, a local + /// member's loaded now ([`RepoReadOps::search_hits`]), one batch per + /// member, a few members at a time. An id a member no longer finds (its + /// last matching version was deleted since) is left out. A member that + /// fails to load its hits fails the search: it answered for those ids a + /// moment ago, and a page left short would read as the last one. + async fn page_hits( + &self, + graph: &GroupGraph, + member_ops: &[Arc], + searches: &mut [MemberSearch], + params: &SearchParams, + page: Vec<(usize, usize)>, + ) -> Result, RepoError> { + use futures_util::{StreamExt, TryStreamExt}; + + let mut local: BTreeMap> = BTreeMap::new(); + for &(member, position) in &page { + if !searches[member].upstream { + local + .entry(member) + .or_default() + .push(searches[member].ids[position].clone()); + } + } + let reads: Vec<_> = local + .into_iter() + .map(|(member, ids)| { + load_hits( + &member_ops[member], + &graph.leaves[member], + params, + member, + ids, + ) + }) + .collect(); + let loaded: Vec<(usize, Vec)> = futures_util::stream::iter(reads) + .buffered(MEMBER_SEARCHES) + .try_collect() + .await?; + let mut loaded: HashMap<(usize, String), SearchHit> = loaded + .into_iter() + .flat_map(|(member, hits)| { + hits.into_iter() + .map(move |hit| ((member, self.lower_id(&hit.original_id)), hit)) + }) + .collect(); + Ok(page + .into_iter() + .filter_map(|(member, position)| { + let search = &mut searches[member]; + let hit = if search.upstream { + std::mem::take(&mut search.hits[position]) + } else { + loaded.remove(&(member, self.lower_id(&search.ids[position])))? + }; + Some(RankedHit { member, hit }) + }) + .collect()) + } + /// Merge the members' search hits: each id once, from the /// highest-priority member that serves it among the hits fetched. The /// ids of local members come first, by id; then those of the upstream @@ -1008,26 +1098,23 @@ impl GroupRepo { for (member, search) in searches.iter().enumerate() { let leaf_id = graph.leaves[member].id; let keys = search - .hits + .ids .iter() .enumerate() - .map(|(position, hit)| { - let rank = graph.leaf_rank(&hit.original_id, leaf_id)?; - let lower = self.lower_id(&hit.original_id); + .map(|(position, id)| { + let rank = graph.leaf_rank(id, leaf_id)?; + let lower = self.lower_id(id); let leaves = serving.entry(lower.clone()).or_insert_with(|| { let guards = IdGuards { lower_id: &lower, owned_in: owned_in.get(&lower).unwrap_or(&no_owner), }; - graph - .serving_leaves(&hit.original_id, &guards) - .into_iter() - .collect() + graph.serving_leaves(id, &guards).into_iter().collect() }); if !leaves.contains(&leaf_id) { return None; } - let key = crate::format::lower_invariant(&hit.original_id); + let key = crate::format::lower_invariant(id); let winner = winners .entry(key.clone()) .or_insert((rank, member, position)); @@ -1066,12 +1153,12 @@ impl GroupRepo { let settled_rounds = upstream .iter() .filter(|member| searches[**member].growable()) - .map(|member| searches[*member].hits.len()) + .map(|member| searches[*member].ids.len()) .min() .unwrap_or(usize::MAX); let rounds = upstream .iter() - .map(|member| searches[*member].hits.len()) + .map(|member| searches[*member].ids.len()) .max() .unwrap_or(0); let mut settled = order.len(); @@ -1097,13 +1184,13 @@ impl GroupRepo { .iter() .map(|&member| &searches[member]) .zip(upstream.iter().map(|&member| credited[member])) - .filter(|(search, _)| search.incomplete() && !search.hits.is_empty()) + .filter(|(search, _)| search.incomplete() && !search.ids.is_empty()) .map(|(search, credited)| { let more = search .total .min(SearchParams::MAX_TAKE) - .saturating_sub(search.hits.len()); - more.saturating_mul(credited) / search.hits.len() + .saturating_sub(search.ids.len()); + more.saturating_mul(credited) / search.ids.len() }) .fold(0_usize, usize::saturating_add); // A search with more hits than it returned may still add ids, though @@ -1492,8 +1579,52 @@ fn upstream_window(wanted: usize) -> usize { .min(SearchParams::MAX_TAKE) } -/// Search one member for a group: its first `window` hits, or its first -/// [`SearchParams::MAX_TAKE`] without one. +/// How many members a group searches at a time. +const MEMBER_SEARCHES: usize = 4; + +/// Whether a group search reads `leaf` through its upstream's search: an +/// online proxy, when `params` include the upstream. Every other member, +/// an offline proxy included, is searched locally (its own packages or its +/// cache). +fn searches_upstream(params: &SearchParams, leaf: &Repository) -> bool { + params.include_upstream && leaf.kind() == RepoKind::Proxy && leaf.online +} + +/// A group's first search of one member: a local member's ids, all of them; +/// a proxy searching its upstream for the window `wanted` hits need. +async fn first_search( + ops: &Arc, + leaf: &Repository, + params: &SearchParams, + filter: &PackageFilter, + wanted: usize, +) -> Result { + if !searches_upstream(params, leaf) { + let ids = ops.search_ids(&leaf.name, params, filter).await?; + return Ok(MemberSearch::local(ids)); + } + // A filter may hide most of the upstream's hits, and the proxy then + // counts only those it returns: it is asked for them all. + let window = matches!(filter, PackageFilter::AllowAll).then(|| upstream_window(wanted)); + let result = search_member(ops, leaf, params, window, filter).await?; + Ok(MemberSearch::upstream(window, result)) +} + +/// The hits of `ids` a local `member` won on the final page +/// ([`RepoReadOps::search_hits`]), tagged with the member. +async fn load_hits( + ops: &Arc, + leaf: &Repository, + params: &SearchParams, + member: usize, + ids: Vec, +) -> Result<(usize, Vec), RepoError> { + let hits = ops.search_hits(&leaf.name, params, &ids).await?; + Ok((member, hits)) +} + +/// Search one member's upstream for a group: its first `window` hits, or its +/// first [`SearchParams::MAX_TAKE`] without one. async fn search_member( ops: &Arc, leaf: &Repository, @@ -1517,7 +1648,11 @@ struct MemberSearch { /// The window of an upstream search the group bounds; `None` when the /// member was asked for all its hits, or can be asked for no more. window: Option, - /// The member's first hits, in its order. + /// The ids of the member's first hits (a local member's: of all its + /// hits), in its order. + ids: Vec, + /// An upstream search's hits, one per id. A local member's are loaded + /// for the final page only ([`GroupRepo::page_hits`]). hits: Vec, /// The member's total. total: usize, @@ -1527,10 +1662,28 @@ struct MemberSearch { } impl MemberSearch { - fn new(upstream: bool, window: Option, result: SearchResult) -> Self { + /// A local member's ids: all of them. + fn local(ids: Vec) -> Self { MemberSearch { - upstream, + upstream: false, + window: None, + total: ids.len(), + ids, + hits: Vec::new(), + stalled: false, + } + } + + /// An upstream search's first hits, for `window`. + fn upstream(window: Option, result: SearchResult) -> Self { + MemberSearch { + upstream: true, window, + ids: result + .hits + .iter() + .map(|hit| hit.original_id.clone()) + .collect(), hits: result.hits, total: result.total_hits, stalled: false, @@ -1542,6 +1695,7 @@ impl MemberSearch { MemberSearch { upstream, window: None, + ids: Vec::new(), hits: Vec::new(), total: 0, stalled: false, @@ -1552,7 +1706,7 @@ impl MemberSearch { fn incomplete(&self) -> bool { self.window .is_some_and(|window| window < SearchParams::MAX_TAKE) - && self.hits.len() < self.total + && self.ids.len() < self.total } /// Whether the member is to be asked for a larger window. @@ -1563,12 +1717,12 @@ impl MemberSearch { /// Whether `result`, for a larger window, extends the hits returned so /// far: the same ids first, in the same order. fn extended_by(&self, result: &SearchResult) -> bool { - result.hits.len() >= self.hits.len() + result.hits.len() >= self.ids.len() && self - .hits + .ids .iter() .zip(&result.hits) - .all(|(had, got)| had.original_id == got.original_id) + .all(|(had, got)| *had == got.original_id) } } diff --git a/crates/core/src/repo/hosted.rs b/crates/core/src/repo/hosted.rs index 7e224cfc..3488023d 100644 --- a/crates/core/src/repo/hosted.rs +++ b/crates/core/src/repo/hosted.rs @@ -93,8 +93,9 @@ impl HostedRepo { /// before paging: the ids that match are listed once, filtered by the /// selector, counted and paged, and only the page's versions are loaded. /// With `enrich`, the returned page carries manifests and download totals - /// ([`Self::enrich_hits`]); a group searches its members without and - /// enriches only its own final page. + /// ([`Self::enrich_hits`]). A group reads the two halves apart: the ids + /// ([`Self::search_names`]), then the hits of its own page + /// ([`Self::hits_of`]), which it enriches. async fn search_filtered( &self, repo: &str, @@ -103,10 +104,32 @@ impl HostedRepo { enrich: bool, ) -> Result { let repo_id = self.repo_id(repo).await?; + // Page over ids, not versions: versions are loaded for the returned + // page only. + let names = self.search_names(repo_id, params, filter).await?; + let total_hits = names.len(); + let page: Vec = names + .into_iter() + .skip(params.skip.min(total_hits)) + .take(params.effective_take()) + .collect(); + let mut hits = self.hits_of(repo_id, params, &page).await?; + if enrich { + self.enrich_hits(repo_id, &mut hits).await?; + } + Ok(SearchResult { total_hits, hits }) + } + + /// The ids (stored, lowercase) of every package a search for `params` + /// finds, in byte order. The filters run in SQL, the selector and the + /// query's terms on the (much shorter) id list. + async fn search_names( + &self, + repo_id: i64, + params: &SearchParams, + filter: &PackageFilter, + ) -> Result, RepoError> { let query = SearchQuery::parse(¶ms.q); - // Page over ids, not versions: the filters run in SQL, the selector - // and the query's terms on the (much shorter) id list, and versions - // are loaded for the returned page only. let mut names: Vec = component_dao::search_names( &self.pool, repo_id, @@ -121,19 +144,24 @@ impl HostedRepo { // Byte order, as group results are ordered (SQL orders by the // database collation, which may weigh punctuation differently). names.sort_unstable(); - let total_hits = names.len(); - let page: Vec = names - .into_iter() - .skip(params.skip.min(total_hits)) - .take(params.effective_take()) - .collect(); + Ok(names) + } + /// The unenriched search hits of `page` (ids [`Self::search_names`] + /// returned for `params`), in its order: each id's matching versions and + /// its casing. An id left without a matching version is skipped. + async fn hits_of( + &self, + repo_id: i64, + params: &SearchParams, + page: &[String], + ) -> Result, RepoError> { use std::collections::HashMap; let mut by_id: HashMap> = HashMap::new(); for c in component_dao::search_versions( &self.pool, repo_id, - &page, + page, params.prerelease, params.include_semver2, ) @@ -142,7 +170,7 @@ impl HostedRepo { by_id.entry(c.name.clone()).or_default().push(c); } let mut hits: Vec = Vec::with_capacity(page.len()); - for name in &page { + for name in page { let Some(comps) = by_id.remove(name) else { // Changed between the two queries. continue; @@ -165,10 +193,7 @@ impl HostedRepo { ..Default::default() }); } - if enrich { - self.enrich_hits(repo_id, &mut hits).await?; - } - Ok(SearchResult { total_hits, hits }) + Ok(hits) } /// Attach the latest version's manifest and the download total to each @@ -531,6 +556,29 @@ impl RepoReadOps for HostedRepo { let repo_id = self.repo_id(repo).await?; self.enrich_hits(repo_id, hits).await } + + async fn search_ids( + &self, + repo: &str, + params: &SearchParams, + filter: &PackageFilter, + ) -> Result, RepoError> { + let repo_id = self.repo_id(repo).await?; + self.search_names(repo_id, params, filter).await + } + + async fn search_hits( + &self, + repo: &str, + params: &SearchParams, + ids: &[String], + ) -> Result, RepoError> { + if ids.is_empty() { + return Ok(Vec::new()); + } + let repo_id = self.repo_id(repo).await?; + self.hits_of(repo_id, params, ids).await + } } /// Storage-backed hosted **write** facade: publish, redeploy, delete, relist. diff --git a/crates/core/src/repo/mod.rs b/crates/core/src/repo/mod.rs index baf173b9..510419ad 100644 --- a/crates/core/src/repo/mod.rs +++ b/crates/core/src/repo/mod.rs @@ -697,10 +697,10 @@ pub trait RepoReadOps: Send + Sync { /// the hits carry ids and versions, but [`SearchHit::nuspec_bytes`] and /// [`SearchHit::total_downloads`] may be left unset. /// - /// A group asks its local members for up to [`SearchParams::MAX_TAKE`] - /// hits, and a proxy that searches its upstream for the leading window of - /// its hits the client's page needs; it merges and counts them, and - /// renders only the client's page (see [`group`]). + /// A group asks a proxy that searches its upstream for the leading window + /// of its hits the client's page needs (its local members for their ids, + /// [`Self::search_ids`]); it merges and counts them, and renders only the + /// client's page (see [`group`]). /// It therefore searches its members through this method and enriches just /// the final page via [`Self::enrich_search_hits`], so manifest reads (S3) /// are bounded by the client's `take` rather than by every member's corpus. @@ -727,6 +727,84 @@ pub trait RepoReadOps: Send + Sync { ) -> Result<(), RepoError> { Ok(()) } + + /// The id of every hit of a search of what `repo` holds itself: the hits + /// [`Self::search_unenriched`] finds for `params` without the upstream + /// (whatever `params.include_upstream`, `skip` and `take` say), in its + /// order. The hosted and proxy backends return the stored (lowercase) + /// ids, in byte order. + /// + /// A group pages its local members' ids rather than their hits, then + /// loads the hits of its page alone, each through the member that won it + /// ([`Self::search_hits`]): the versions of one page, not of every + /// member's leading hits (see [`group`]). + /// + /// The default pages through [`Self::search_unenriched`]: correct for any + /// implementation, merely not cheaper. + async fn search_ids( + &self, + repo: &str, + params: &SearchParams, + filter: &PackageFilter, + ) -> Result, RepoError> { + Ok(all_local_hits(self, repo, params, filter) + .await? + .into_iter() + .map(|hit| hit.original_id) + .collect()) + } + + /// The unenriched hits ([`Self::search_unenriched`]) of `ids`, which + /// [`Self::search_ids`] returned for the same `params`, in the order of + /// `ids`. An id that is no hit any more (its last matching version was + /// deleted since) is left out. + /// + /// The default searches again ([`Self::search_ids`]' default). + async fn search_hits( + &self, + repo: &str, + params: &SearchParams, + ids: &[String], + ) -> Result, RepoError> { + let mut hits: std::collections::HashMap = + all_local_hits(self, repo, params, &PackageFilter::AllowAll) + .await? + .into_iter() + .map(|hit| (hit.original_id.clone(), hit)) + .collect(); + Ok(ids.iter().filter_map(|id| hits.remove(id)).collect()) + } +} + +/// Every unenriched hit of a search of what `repo` holds itself, page by +/// page: the defaults of [`RepoReadOps::search_ids`] and +/// [`RepoReadOps::search_hits`]. +async fn all_local_hits( + ops: &R, + repo: &str, + params: &SearchParams, + filter: &PackageFilter, +) -> Result, RepoError> { + let mut hits = Vec::new(); + loop { + let page = ops + .search_unenriched( + repo, + &SearchParams { + skip: hits.len(), + take: SearchParams::MAX_TAKE, + include_upstream: false, + ..params.clone() + }, + filter, + ) + .await?; + let returned = page.hits.len(); + hits.extend(page.hits); + if returned == 0 || hits.len() >= page.total_hits { + return Ok(hits); + } + } } /// A package-filtering decorator around any repository read implementation. @@ -899,6 +977,28 @@ impl RepoReadOps for PackageFilteredReadOps { // the predicate; enrichment adds no package identities. self.inner.enrich_search_hits(repo, hits).await } + + async fn search_ids( + &self, + repo: &str, + params: &SearchParams, + filter: &PackageFilter, + ) -> Result, RepoError> { + let combined = self.filter.clone().intersect(filter.clone()); + self.inner.search_ids(repo, params, &combined).await + } + + async fn search_hits( + &self, + repo: &str, + params: &SearchParams, + ids: &[String], + ) -> Result, RepoError> { + // The ids came out of the filtered search above; an id named any other + // way is refused all the same. + let ids: Vec = ids.iter().filter(|id| self.allows(id)).cloned().collect(); + self.inner.search_hits(repo, params, &ids).await + } } /// A boxed async byte source, used by [`RepoWriteOps::store_upload`] so the diff --git a/crates/core/src/repo/proxy.rs b/crates/core/src/repo/proxy.rs index 67e02034..d42a9e0a 100644 --- a/crates/core/src/repo/proxy.rs +++ b/crates/core/src/repo/proxy.rs @@ -2259,8 +2259,9 @@ impl ProxyRepo { /// versions are loaded for the requested page alone, so an empty query /// over a large proxy cache does not load every cached version. /// With `enrich`, the returned page carries cached manifests - /// ([`Self::enrich_hits`]); a group searches its members without and - /// enriches only its own final page. + /// ([`Self::enrich_hits`]). A group reads the two halves apart: the ids + /// ([`Self::search_names`]), then the hits of its own page + /// ([`Self::hits_of`]), which it enriches. async fn search_filtered( &self, repo: &str, @@ -2269,11 +2270,42 @@ impl ProxyRepo { enrich: bool, ) -> Result { let repo_row = self.proxy_repo(repo).await?; + let bounds = self.listable_bounds(&repo_row).await?; + let names = self + .search_names(&repo_row, &bounds, params, filter) + .await?; + let total_hits = names.len(); + let page: Vec = names + .into_iter() + .skip(params.skip.min(total_hits)) + .take(params.effective_take()) + .collect(); + let mut hits = self.hits_of(&repo_row, &bounds, params, &page).await?; + if enrich { + self.enrich_hits(repo_row.id, &mut hits).await; + } + Ok(SearchResult { total_hits, hits }) + } + + /// The `listable` conditions of a cache search, as of now + /// ([`ListableBounds`]). + async fn listable_bounds(&self, repo_row: &Repository) -> Result { let now = self.db_now().await?; - // The `listable` conditions as bounds: past the delay window, and no - // not-found marker younger than the TTL. - let seen_before = now - Duration::seconds(i64::from(repo_row.package_delay_seconds)); - let marked_after = now - Self::not_found_ttl(&repo_row); + Ok(ListableBounds { + seen_before: now - Duration::seconds(i64::from(repo_row.package_delay_seconds)), + marked_after: now - Self::not_found_ttl(repo_row), + }) + } + + /// The ids (stored, lowercase) of every cached package a search for + /// `params` finds, in byte order. + async fn search_names( + &self, + repo_row: &Repository, + bounds: &ListableBounds, + params: &SearchParams, + filter: &PackageFilter, + ) -> Result, RepoError> { let query = SearchQuery::parse(¶ms.q); let q = query.prefilter().unwrap_or("").to_string(); // SemVer 2.0.0 for legacy clients: a `.` after a `-`. `version` is the @@ -2294,8 +2326,8 @@ impl ProxyRepo { "#, repo_row.id, q, - seen_before, - marked_after, + bounds.seen_before, + bounds.marked_after, params.prerelease, params.include_semver2, ) @@ -2308,13 +2340,19 @@ impl ProxyRepo { // Byte order, as hosted and group results are ordered (SQL orders by // the database collation, which may weigh punctuation differently). names.sort_unstable(); - let total_hits = names.len(); - let page: Vec = names - .into_iter() - .skip(params.skip.min(total_hits)) - .take(params.effective_take()) - .collect(); + Ok(names) + } + /// The unenriched search hits of `page` (ids [`Self::search_names`] + /// returned for `params`), in its order: each id's listable versions and + /// its casing. An id left without one is skipped. + async fn hits_of( + &self, + repo_row: &Repository, + bounds: &ListableBounds, + params: &SearchParams, + page: &[String], + ) -> Result, RepoError> { let rows = sqlx::query_as!( forge_db::model::ProxyCachedVersion, r#" @@ -2329,9 +2367,9 @@ impl ProxyRepo { AND ($6 OR version !~ '-.*[.]') "#, repo_row.id, - &page, - seen_before, - marked_after, + page, + bounds.seen_before, + bounds.marked_after, params.prerelease, params.include_semver2, ) @@ -2349,12 +2387,12 @@ impl ProxyRepo { // The id as the package spells it (rows hold the lowercased one). let original_ids: HashMap = - proxy_dao::original_ids(&self.pool, repo_row.id, &page) + proxy_dao::original_ids(&self.pool, repo_row.id, page) .await? .into_iter() .collect(); let mut hits: Vec = Vec::with_capacity(page.len()); - for name in &page { + for name in page { let Some(mut versions) = by_id.remove(name) else { // Changed between the two queries. continue; @@ -2375,10 +2413,7 @@ impl ProxyRepo { ..Default::default() }); } - if enrich { - self.enrich_hits(repo_row.id, &mut hits).await; - } - Ok(SearchResult { total_hits, hits }) + Ok(hits) } /// Attach each hit's latest manifest. Manifests are known for cached @@ -2916,6 +2951,42 @@ impl RepoReadOps for ProxyRepo { self.enrich_hits(repo_row.id, hits).await; Ok(()) } + + /// The cached packages' ids: a group asks a proxy for them only where it + /// searches the cache (without the upstream, or while offline). + async fn search_ids( + &self, + repo: &str, + params: &SearchParams, + filter: &PackageFilter, + ) -> Result, RepoError> { + let repo_row = self.proxy_repo(repo).await?; + let bounds = self.listable_bounds(&repo_row).await?; + self.search_names(&repo_row, &bounds, params, filter).await + } + + async fn search_hits( + &self, + repo: &str, + params: &SearchParams, + ids: &[String], + ) -> Result, RepoError> { + if ids.is_empty() { + return Ok(Vec::new()); + } + let repo_row = self.proxy_repo(repo).await?; + let bounds = self.listable_bounds(&repo_row).await?; + self.hits_of(&repo_row, &bounds, params, ids).await + } +} + +/// The `listable` conditions of a search of a proxy's cache as bounds, taken +/// at one instant: a version is listed once first seen at or before +/// `seen_before` (past the package delay), unless a not-found marker set +/// after `marked_after` (younger than the TTL) hides it. +struct ListableBounds { + seen_before: DateTime, + marked_after: DateTime, } /// One page of an upstream's search results, reduced to what a search hit diff --git a/crates/core/tests/group_read.rs b/crates/core/tests/group_read.rs index 3ee648ce..9a7abd02 100644 --- a/crates/core/tests/group_read.rs +++ b/crates/core/tests/group_read.rs @@ -2826,8 +2826,9 @@ async fn upstream_searches_are_merged_in_turn() { assert_eq!(totals[0], 100); } -/// An offline proxy searches its cache: the group asks it for all its hits -/// and counts them exactly, as it does a hosted member's. +/// An offline proxy searches its cache: the group asks it for all its ids +/// and counts them exactly, as it does a hosted member's, then for the hits +/// of its page. (The stub answers both through its search, unwindowed.) #[tokio::test] async fn an_offline_proxy_is_searched_like_a_local_member() { let pool = common::pool().await; @@ -2851,7 +2852,10 @@ async fn an_offline_proxy_is_searched_like_a_local_member() { let (pages, totals) = search_pages(&g, &group.name, "Cached", 20, 1).await; assert_eq!(pages[0], ids("Cached", 0..20)); assert_eq!(totals[0], 60); - assert_eq!(upstream.takes(), [SearchParams::MAX_TAKE]); + assert_eq!( + upstream.takes(), + [SearchParams::MAX_TAKE, SearchParams::MAX_TAKE] + ); } /// Only the upstream search whose hits hold the page back is widened: one @@ -3007,3 +3011,339 @@ async fn an_upstream_window_the_proxy_emptied_keeps_the_total_past_the_page() { assert_eq!(totals[0], 41, "one past the 40 local ids"); assert_eq!(pages[2], ids("Delayed", 32..52)); } + +// ── local members merged on ids (#536) ────────────────────────────────────── + +/// One search call a [`Recording`] member saw, with the member. +#[derive(Debug, Clone, PartialEq, Eq)] +enum SearchCall { + /// `search_ids`. + Ids(String), + /// `search_hits`, for these ids. + Hits(String, Vec), + /// A search of hits with their versions (`search`, `search_unenriched`). + Search(String), +} + +/// A member's read ops that record every search call before passing it on. +struct Recording { + inner: Arc, + calls: Arc>>, +} + +#[async_trait] +impl RepoReadOps for Recording { + async fn list_versions( + &self, + repo: &str, + id: &str, + ) -> Result, RepoError> { + self.inner.list_versions(repo, id).await + } + + async fn resolve_nupkg( + &self, + repo: &str, + id: &str, + version: &str, + ) -> Result, RepoError> { + self.inner.resolve_nupkg(repo, id, version).await + } + + async fn resolve_nuspec( + &self, + repo: &str, + id: &str, + version: &str, + ) -> Result>, RepoError> { + self.inner.resolve_nuspec(repo, id, version).await + } + + async fn registration( + &self, + repo: &str, + id: &str, + ) -> Result, RepoError> { + self.inner.registration(repo, id).await + } + + async fn search( + &self, + repo: &str, + params: &SearchParams, + ) -> Result { + self.calls + .lock() + .unwrap() + .push(SearchCall::Search(repo.into())); + self.inner.search(repo, params).await + } + + async fn search_unenriched( + &self, + repo: &str, + params: &SearchParams, + filter: &forge_core::authz::PackageFilter, + ) -> Result { + self.calls + .lock() + .unwrap() + .push(SearchCall::Search(repo.into())); + self.inner.search_unenriched(repo, params, filter).await + } + + async fn enrich_search_hits( + &self, + repo: &str, + hits: &mut [forge_core::repo::SearchHit], + ) -> Result<(), RepoError> { + self.inner.enrich_search_hits(repo, hits).await + } + + async fn search_ids( + &self, + repo: &str, + params: &SearchParams, + filter: &forge_core::authz::PackageFilter, + ) -> Result, RepoError> { + self.calls + .lock() + .unwrap() + .push(SearchCall::Ids(repo.into())); + self.inner.search_ids(repo, params, filter).await + } + + async fn search_hits( + &self, + repo: &str, + params: &SearchParams, + ids: &[String], + ) -> Result, RepoError> { + self.calls + .lock() + .unwrap() + .push(SearchCall::Hits(repo.into(), ids.to_vec())); + self.inner.search_hits(repo, params, ids).await + } +} + +/// Hosted members, recorded ([`Recording`]); proxy-kind members are +/// [`RankedUpstream`]s by id, searched locally while offline. +struct RecordingFactory { + hosted: HostedFactory, + upstreams: HashMap>, + calls: Arc>>, +} + +impl MemberOpsFactory for RecordingFactory { + fn ops_for(&self, member: &Repository) -> Result, RepoError> { + if let Some(upstream) = self.upstreams.get(&member.id) { + return Ok(upstream.clone()); + } + Ok(Arc::new(Recording { + inner: self.hosted.ops_for(member)?, + calls: self.calls.clone(), + })) + } +} + +fn recording_group_repo( + pool: &PgPool, + principal: Principal, + upstreams: &[(&Repository, &Arc)], +) -> (GroupRepo, Arc>>) { + let calls = Arc::new(std::sync::Mutex::new(Vec::new())); + let factory = Arc::new(RecordingFactory { + hosted: HostedFactory { + pool: pool.clone(), + blobs: StubBlobs::default(), + }, + upstreams: upstreams + .iter() + .map(|(proxy, upstream)| (proxy.id, Arc::clone(upstream))) + .collect(), + calls: calls.clone(), + }); + let authz = Authz::new(pool.clone(), Cache::l1_only(16)); + let g = GroupRepo::new(pool.clone(), principal, factory, authz, TEST_MAX_DEPTH) + .with_format(Arc::new(nuget())); + (g, calls) +} + +/// Seed `.` (`n` zero-padded to four digits) for each `n` of +/// `range` into hosted `repo_id`, each in `versions`, `listed` or not: bare +/// component rows, all a search reads. +async fn seed_ids( + pool: &PgPool, + repo_id: i64, + prefix: &str, + range: std::ops::Range, + versions: &[&str], + listed: bool, +) { + let versions: Vec = versions.iter().map(|v| v.to_string()).collect(); + sqlx::query( + "INSERT INTO component (repository_id, name, original_id, version, version_original, listed) + SELECT $1, lower(id), id, v, v, $6 + FROM (SELECT $2 || '.' || lpad(n::text, 4, '0') AS id + FROM generate_series($3, $4 - 1) AS n) AS ids, + unnest($5::text[]) AS v", + ) + .bind(repo_id) + .bind(prefix) + .bind(range.start) + .bind(range.end) + .bind(&versions) + .bind(listed) + .execute(pool) + .await + .unwrap(); +} + +fn padded(prefix: &str, range: std::ops::Range) -> Vec { + range.map(|n| format!("{prefix}.{n:04}")).collect() +} + +/// A group pages its local members' ids, and loads versions and casing for +/// its page alone, each hit through the member that won it (#536). It reads +/// past each member's first 1,000 ids, and counts every merged id. +#[tokio::test] +async fn a_group_pages_local_ids_and_loads_only_its_page() { + let pool = common::pool().await; + let high = seed_member(&pool, false).await; + let low = seed_member(&pool, false).await; + let group = seed_group(&pool).await; + let p = format!("Local{}", common::unique()); + // high: 0000..1199 in three versions; low: 1100..1399 in two, so + // 1100..1199 are shared and high's copy wins them. + seed_ids( + &pool, + high.id, + &p, + 0..1200, + &["1.0.0", "1.1.0", "1.2.0"], + true, + ) + .await; + seed_ids(&pool, low.id, &p, 1100..1400, &["2.0.0", "2.1.0"], true).await; + repository_dao::add_group_member(&pool, group.id, high.id, 100) + .await + .unwrap(); + repository_dao::add_group_member(&pool, group.id, low.id, 10) + .await + .unwrap(); + let (g, calls) = recording_group_repo(&pool, user(&pool).await, &[]); + + let search = |skip: usize| { + let (g, group, q) = (&g, &group.name, p.clone()); + async move { + g.search( + group, + &SearchParams { + q, + skip, + take: 20, + include_upstream: false, + ..SearchParams::default() + }, + ) + .await + .unwrap() + } + }; + + let result = search(1190).await; + assert_eq!(result.total_hits, 1400, "every merged id counts"); + let page: Vec<&str> = result.hits.iter().map(|h| h.original_id.as_str()).collect(); + assert_eq!( + page, + padded(&p, 1190..1210), + "ids past 1,000, with their casing" + ); + for hit in &result.hits { + let n: usize = hit.original_id.rsplit('.').next().unwrap().parse().unwrap(); + let (versions, latest) = if n < 1200 { + (vec!["1.0.0", "1.1.0", "1.2.0"], "1.2.0") + } else { + (vec!["2.0.0", "2.1.0"], "2.1.0") + }; + let got: Vec<&str> = hit.versions.iter().map(|v| v.version.as_str()).collect(); + assert_eq!( + got, versions, + "{}: the winning member's versions", + hit.original_id + ); + assert_eq!(hit.latest_version, latest); + } + let lower = |range| -> Vec { + padded(&p, range) + .into_iter() + .map(|id| id.to_lowercase()) + .collect() + }; + let mut seen = calls.lock().unwrap().clone(); + seen.sort_by_key(|call| format!("{call:?}")); + let mut expected = vec![ + SearchCall::Ids(high.name.clone()), + SearchCall::Ids(low.name.clone()), + SearchCall::Hits(high.name.clone(), lower(1190..1200)), + SearchCall::Hits(low.name.clone(), lower(1200..1210)), + ]; + expected.sort_by_key(|call| format!("{call:?}")); + assert_eq!( + seen, expected, + "each member lists its ids once, and loads the hits it won on the page, no others" + ); + + let last = search(1390).await; + assert_eq!(last.total_hits, 1400); + let page: Vec<&str> = last.hits.iter().map(|h| h.original_id.as_str()).collect(); + assert_eq!(page, padded(&p, 1390..1400), "the last page is reached"); +} + +/// The total of a group over a local member past its first 1,000 ids is +/// exact under the dependency-confusion guard: ids an owner holds (here +/// unlisted only, so it lists none of them) are taken off a cache's hits, +/// wherever they sort. +#[tokio::test] +async fn a_guard_takes_ids_past_the_first_thousand_off_the_total() { + let pool = common::pool().await; + let hosted = seed_member(&pool, false).await; + let proxy = seed_proxy_member(&pool).await; + let group = seed_group(&pool).await; + repository_dao::add_group_member(&pool, group.id, hosted.id, 100) + .await + .unwrap(); + repository_dao::add_group_member(&pool, group.id, proxy.id, 10) + .await + .unwrap(); + sqlx::query("UPDATE repository SET online = false WHERE id = $1") + .bind(proxy.id) + .execute(&pool) + .await + .unwrap(); + let p = format!("Owned{}", common::unique()); + seed_ids(&pool, hosted.id, &p, 1050..1060, &["1.0.0"], false).await; + let cache = RankedUpstream::new(padded(&p, 0..1100), 1100); + let (g, _) = recording_group_repo(&pool, user(&pool).await, &[(&proxy, &cache)]); + + let result = g + .search( + &group.name, + &SearchParams { + q: p.clone(), + skip: 1040, + take: 20, + ..SearchParams::default() + }, + ) + .await + .unwrap(); + assert_eq!(result.total_hits, 1090); + let page: Vec = result.hits.into_iter().map(|h| h.original_id).collect(); + assert_eq!( + page, + [padded(&p, 1040..1050), padded(&p, 1060..1070)].concat(), + "the owned ids are left out" + ); +} diff --git a/crates/server/tests/browse.rs b/crates/server/tests/browse.rs index c5acabbe..013fe0b7 100644 --- a/crates/server/tests/browse.rs +++ b/crates/server/tests/browse.rs @@ -494,6 +494,117 @@ async fn browse_group_merges_members() { let _ = lower; } +/// A group's browse reaches every id its members hold, past each member's +/// first 1,000, and counts them all (#536): here a proxy whose cache lists +/// 1,500 ids (seeded in SQL) and a hosted member that owns one of them. +#[tokio::test] +async fn a_group_browses_every_cached_id_of_its_proxy() { + let srv = common::start().await; + let (u, p) = srv.admin_basic(); + let client = srv.client(); + let suffix = common::unique(); + let hosted = format!("bigh{suffix}"); + let proxy = format!("bigp{suffix}"); + let group = format!("bigg{suffix}"); + let prefix = format!("Big{suffix}"); + let id = |n: usize| format!("{prefix}.P{n:04}"); + + publish_one(&srv, &hosted, &id(1490), "9.0.0", false).await; + let resp = client + .post(format!("{}/admin/repositories", srv.base_url)) + .basic_auth(&u, Some(&p)) + .json(&json!({"name": proxy, "kind": "proxy", + "remote_url": "https://feed.example/v3/index.json"})) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 201); + let proxy_id = resp.json::().await.unwrap()["id"].as_i64().unwrap(); + sqlx::query( + "INSERT INTO proxy_cached_version + (repository_id, name, version, first_seen_at, original_id) + SELECT $1, lower(id), v, now() - interval '1 day', id + FROM (SELECT $2 || '.P' || lpad(n::text, 4, '0') AS id + FROM generate_series(0, 1499) AS n) AS ids, + unnest(ARRAY['1.0.0', '1.1.0', '1.2.0']) AS v", + ) + .bind(proxy_id) + .bind(&prefix) + .execute(&srv.pool) + .await + .unwrap(); + let resp = client + .post(format!("{}/admin/repositories", srv.base_url)) + .basic_auth(&u, Some(&p)) + .json(&json!({"name": group, "kind": "group"})) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 201); + for (member, priority) in [(&hosted, 100), (&proxy, 10)] { + let resp = client + .post(format!( + "{}/admin/repositories/{group}/members", + srv.base_url + )) + .basic_auth(&u, Some(&p)) + .json(&json!({"member": member, "priority": priority})) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 201); + } + + let page = |query: String| { + let (client, srv, group) = (&client, &srv, &group); + let (u, p) = (u.clone(), p.clone()); + async move { + let resp = client + .get(format!( + "{}/browse/packages?repo={group}&{query}", + srv.base_url + )) + .basic_auth(&u, Some(&p)) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + let body: Value = resp.json().await.unwrap(); + let items: Vec<(String, String)> = body["items"] + .as_array() + .unwrap() + .iter() + .map(|item| { + ( + item["original_id"].as_str().unwrap().to_string(), + item["latest_version"].as_str().unwrap().to_string(), + ) + }) + .collect(); + (body["total"].as_u64().unwrap(), items) + } + }; + let expected = |range: std::ops::Range| -> Vec<(String, String)> { + range + .map(|n| { + let version = if n == 1490 { "9.0.0" } else { "1.2.0" }; + (id(n), version.to_string()) + }) + .collect() + }; + + assert_eq!(page("take=20".into()).await, (1500, expected(0..20))); + assert_eq!( + page("skip=1480&take=20".into()).await, + (1500, expected(1480..1500)), + "the cache's ids past 1,000, the owned one from the hosted member" + ); + assert_eq!( + page(format!("q={}&take=5", id(1495).to_lowercase())).await, + (1, expected(1495..1496)) + ); +} + /// A NuGet group shows a caller what it could read from its members /// directly, no more (#648): with a member it may not read, a selector on /// the group's grant or one on a member's, a hidden package has no flat diff --git a/docs/changes/2.2.0/536.fixed.md b/docs/changes/2.2.0/536.fixed.md new file mode 100644 index 00000000..4c6ef481 --- /dev/null +++ b/docs/changes/2.2.0/536.fixed.md @@ -0,0 +1,8 @@ +Browsing a NuGet group in the web UI loaded every version of each member's +first 1,000 packages on every page, and listed no package past them: a +member's 1,001st package was reachable only by browsing the member itself, +and the group's total stopped there. A group now lists, counts and pages +the ids of all its members' packages, and loads versions for the packages +on the page alone. Its members are searched a few at a time rather than one +after another. NuGet searches of a group over hosted members, or over an +offline proxy, change the same way.