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.