diff --git a/modules/analysis/src/endpoints/tests/mod.rs b/modules/analysis/src/endpoints/tests/mod.rs index 772c70209..ac888c736 100644 --- a/modules/analysis/src/endpoints/tests/mod.rs +++ b/modules/analysis/src/endpoints/tests/mod.rs @@ -164,6 +164,49 @@ async fn test_quarkus_retrieve_analysis_endpoint( Ok(()) } +/// Verify pagination through the HTTP endpoint returns correct total and limited items. +#[test_context(TrustifyContext)] +#[test(actix_web::test)] +async fn test_pagination_through_endpoint(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + let app = caller(ctx).await?; + ctx.ingest_documents([ + "spdx/quarkus-bom-3.2.11.Final-redhat-00001.json", + "spdx/quarkus-bom-3.2.12.Final-redhat-00002.json", + ]) + .await?; + + // When: request with limit=1, offset=0 + let page1: Value = app + .req(Req { + what: What::Q("spymemcached"), + limit: Some(1), + offset: Some(0), + ..Req::default() + }) + .await?; + + assert_eq!(page1["total"], 2); + assert_eq!(page1["items"].as_array().map(|a| a.len()), Some(1)); + + // When: request page 2 + let page2: Value = app + .req(Req { + what: What::Q("spymemcached"), + limit: Some(1), + offset: Some(1), + ..Req::default() + }) + .await?; + + assert_eq!(page2["total"], 2); + assert_eq!(page2["items"].as_array().map(|a| a.len()), Some(1)); + + // Then: pages contain different items + assert_ne!(page1["items"][0]["sbom_id"], page2["items"][0]["sbom_id"]); + + Ok(()) +} + #[test_context(TrustifyContext)] #[test(actix_web::test)] async fn test_status_endpoint(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { diff --git a/modules/analysis/src/service/load/mod.rs b/modules/analysis/src/service/load/mod.rs index 8a60cd639..724627b7e 100644 --- a/modules/analysis/src/service/load/mod.rs +++ b/modules/analysis/src/service/load/mod.rs @@ -16,12 +16,12 @@ use opentelemetry::KeyValue; use petgraph::{Graph, prelude::NodeIndex}; use sea_orm::{ ColumnTrait, ConnectionTrait, DatabaseBackend, DbErr, EntityOrSelect, EntityTrait, - FromQueryResult, QueryFilter, QuerySelect, QueryTrait, RelationTrait, Statement, + FromQueryResult, QueryFilter, QueryOrder, QuerySelect, QueryTrait, RelationTrait, Statement, }; use sea_query::{JoinType, SelectStatement}; use serde_json::Value; use std::{ - collections::{HashMap, HashSet, hash_map::Entry}, + collections::{BTreeSet, HashMap, HashSet, hash_map::Entry}, fmt::{Debug, Display, Formatter}, hash::Hash, str::FromStr, @@ -449,8 +449,8 @@ impl InnerService { apply_rank(&mut ranked_sboms); log::trace!("ranked sboms: {:?}", TruncatedIter(&ranked_sboms)); - // retrieve only ranked_sboms with rank = 1 - let latest_ids: HashSet<_> = ranked_sboms + // retrieve only ranked_sboms with rank = 1, sorted for deterministic pagination + let latest_ids: BTreeSet<_> = ranked_sboms .into_iter() .filter(|item| item.rank == Some(1)) .map(|item| item.matched_sbom_id) @@ -471,6 +471,7 @@ impl InnerService { ) -> Result)>, Error> { let distinct_sbom_ids = sbom::Entity::find() .filter(sbom::Column::SbomId.in_subquery(subquery)) + .order_by_asc(sbom::Column::SbomId) .select() .all(connection) .await? diff --git a/modules/analysis/src/service/mod.rs b/modules/analysis/src/service/mod.rs index 054104097..bfe35c864 100644 --- a/modules/analysis/src/service/mod.rs +++ b/modules/analysis/src/service/mod.rs @@ -502,6 +502,169 @@ impl AnalysisService { .await } + /// Count nodes matching the query across all graphs, without hydrating them. + fn count_matching_nodes(query: &GraphQuery, graphs: &[(Uuid, Arc)]) -> u64 { + graphs + .iter() + .map(|(_, graph)| { + graph + .node_indices() + .filter(|&i| Self::filter(graph, query, i)) + .count() as u64 + }) + .sum() + } + + /// Collect nodes from the graph, applying pagination at the index level. + /// + /// Flattens matching nodes across ALL graphs into a single sequence, sorts + /// by `(sbom_id, node_id)` for deterministic ordering, then applies skip/take + /// globally. Only the paged subset runs through the expensive `create` closure. + #[instrument(skip(self, create, graphs))] + async fn collect_graph_paged<'a, 'g, F, Fut>( + &self, + query: impl Into> + Debug, + graphs: &'g [(Uuid, Arc)], + concurrency: usize, + offset: usize, + limit: usize, + create: F, + ) -> Result, Error> + where + F: Fn(&'g Graph, NodeIndex, &'g graph::Node) -> Fut + Clone, + Fut: Future>, + { + let query = query.into(); + + let mut matching: Vec<_> = graphs + .iter() + .flat_map(|(_, graph)| { + graph + .node_indices() + .filter(|&i| Self::filter(graph, &query, i)) + .filter_map(|i| graph.node_weight(i).map(|w| (graph.as_ref(), i, w))) + }) + .collect(); + + matching.sort_by(|a, b| { + a.2.sbom_id + .cmp(&b.2.sbom_id) + .then_with(|| a.2.node_id.cmp(&b.2.node_id)) + }); + + let matching: Vec<_> = if limit == 0 { + matching.into_iter().skip(offset).collect() + } else { + matching.into_iter().skip(offset).take(limit).collect() + }; + + stream::iter(matching) + .map(|(graph, node_index, package_node)| { + let create = create.clone(); + create(graph, node_index, package_node) + }) + .buffer_unordered(concurrency) + .try_collect::>() + .await + } + + /// Run graph query with pagination applied before hydration. + #[instrument(skip(self, connection, graphs))] + async fn run_graph_query_paged<'a, C: ConnectionTrait>( + &self, + query: impl Into> + Debug, + options: QueryOptions, + graphs: &[(Uuid, Arc)], + paginated: Paginated, + connection: &C, + ) -> Result, Error> { + let query = query.into(); + let relationships = options.relationships; + tracing::debug!(?relationships, "relations"); + + let total = Self::count_matching_nodes(&query, graphs); + + let limit = paginated.limit as usize; + let offset = paginated.offset as usize; + + if offset as u64 >= total { + return Ok(PaginatedResults { + items: vec![], + total, + }); + } + + let loader = &GraphLoader::new(self.clone()); + + let items = self + .collect_graph_paged( + query, + graphs, + self.concurrency, + offset, + limit, + |graph, node_index, node| { + let graph_cache = self.inner.graph_cache.clone(); + let relationships = relationships.clone(); + async move { + tracing::trace!( + "Discovered node - sbom: {}, node: {}", + node.sbom_id, + node.node_id + ); + + let ancestors = Collector::new( + &graph_cache, + graphs, + node.sbom_id, + graph, + node_index, + Direction::Incoming, + options.ancestors, + &relationships, + connection, + self.concurrency, + loader, + ) + .collect(); + + let descendants = Collector::new( + &graph_cache, + graphs, + node.sbom_id, + graph, + node_index, + Direction::Outgoing, + options.descendants, + &relationships, + connection, + self.concurrency, + loader, + ) + .collect(); + + let (ancestors, descendants) = futures::join!(ancestors, descendants); + let ancestors = ancestors?; + let descendants = descendants?; + + let mut warnings = ancestors.1; + warnings.extend(descendants.1); + + Ok(Node { + base: node.into(), + relationship: None, + ancestors: ancestors.0, + descendants: descendants.0, + warnings, + }) + } + }, + ) + .await?; + + Ok(PaginatedResults { items, total }) + } + #[instrument(skip(self, connection, graphs))] pub async fn run_graph_query<'a, C: ConnectionTrait>( &self, @@ -602,11 +765,8 @@ impl AnalysisService { let options = options.into(); let graphs = self.load_graphs(connection, distinct_sbom_ids).await?; - let components = self - .run_graph_query(query, options, &graphs, connection) - .await?; - - Ok(paginated.paginate_array(&components)) + self.run_graph_query_paged(query, options, &graphs, paginated, connection) + .await } /// locate components, retrieve dependency information @@ -623,11 +783,8 @@ impl AnalysisService { let graphs = self.load_graphs_query(connection, query).await?; - let components = self - .run_graph_query(query, options, &graphs, connection) - .await?; - - Ok(paginated.paginate_array(&components)) + self.run_graph_query_paged(query, options, &graphs, paginated, connection) + .await } pub(crate) async fn load_graphs_query( @@ -649,19 +806,15 @@ impl AnalysisService { let query = query.into(); let options = options.into(); - // load only latest graphs let graphs = self .inner .load_latest_graphs_query(connection, query) .await?; - log::debug!("graph sbom count: {:?}", graphs.len()); + tracing::debug!("graph sbom count: {:?}", graphs.len()); - let components = self - .run_graph_query(query, options, &graphs, connection) - .await?; - - Ok(paginated.paginate_array(&components)) + self.run_graph_query_paged(query, options, &graphs, paginated, connection) + .await } /// check if a node in the graph matches the provided query diff --git a/modules/analysis/src/service/test/mod.rs b/modules/analysis/src/service/test/mod.rs index 0e2c92f6b..ebe1fdb26 100644 --- a/modules/analysis/src/service/test/mod.rs +++ b/modules/analysis/src/service/test/mod.rs @@ -852,3 +852,100 @@ async fn resolve_sbom_cdx_rh_variant_external_node_sbom( Ok(()) } + +/// Verify that pagination limits the number of hydrated nodes. +#[test_context(TrustifyContext)] +#[test(tokio::test)] +async fn test_pagination_limits_hydration(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + ctx.ingest_documents([ + "spdx/quarkus-bom-3.2.11.Final-redhat-00001.json", + "spdx/quarkus-bom-3.2.12.Final-redhat-00002.json", + ]) + .await?; + + let service = AnalysisService::new(AnalysisConfig::default(), ctx.db.clone()); + + // When: request with limit=1 + let result = service + .retrieve( + &Query::q("spymemcached"), + QueryOptions::ancestors(), + Paginated { + offset: 0, + limit: 1, + }, + &ctx.db, + ) + .await?; + + // Then: only 1 item returned but total reflects all matches + assert_eq!(result.items.len(), 1); + assert_eq!(result.total, 2); + + Ok(()) +} + +/// Verify that offset produces different results from different pages. +#[test_context(TrustifyContext)] +#[test(tokio::test)] +async fn test_pagination_offset(ctx: &TrustifyContext) -> Result<(), anyhow::Error> { + ctx.ingest_documents([ + "spdx/quarkus-bom-3.2.11.Final-redhat-00001.json", + "spdx/quarkus-bom-3.2.12.Final-redhat-00002.json", + ]) + .await?; + + let service = AnalysisService::new(AnalysisConfig::default(), ctx.db.clone()); + + // Given: page 1 (offset=0, limit=1) + let page1 = service + .retrieve( + &Query::q("spymemcached"), + QueryOptions::ancestors(), + Paginated { + offset: 0, + limit: 1, + }, + &ctx.db, + ) + .await?; + + // Given: page 2 (offset=1, limit=1) + let page2 = service + .retrieve( + &Query::q("spymemcached"), + QueryOptions::ancestors(), + Paginated { + offset: 1, + limit: 1, + }, + &ctx.db, + ) + .await?; + + // Then: both pages have 1 item each + assert_eq!(page1.items.len(), 1); + assert_eq!(page2.items.len(), 1); + + // Then: the items are different (from different SBOMs) + assert_ne!(&page1.items[0].base.sbom_id, &page2.items[0].base.sbom_id); + + // Given: combined page (offset=0, limit=2) + let combined = service + .retrieve( + &Query::q("spymemcached"), + QueryOptions::ancestors(), + Paginated { + offset: 0, + limit: 2, + }, + &ctx.db, + ) + .await?; + + // Then: combined contains both items + assert_eq!(combined.items.len(), 2); + assert_eq!(combined.total, 2); + + Ok(()) +}