Skip to content

perf(analysis): paginate before hydration to fix component search timeout #2446

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mrrajan wants to merge 3 commits into
base: main
Choose a base branch
from
+402 −21
Open

perf(analysis): paginate before hydration to fix component search timeout #2446

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
633cc67
perf(analysis): paginate before hydration to fix component search tim…
mrrajan Jun 15, 2026
0c2a598
style: fix cargo fmt formatting in endpoint tests
mrrajan Jun 16, 2026
28243b7
refactor(analysis): use BTreeSet for sorted unique SBOM IDs
mrrajan Jun 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 45 additions & 0 deletions modules/analysis/src/endpoints/tests/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,51 @@ 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, total=true
let page1: Value = app
.req(Req {
what: What::Q("spymemcached"),
limit: Some(1),
offset: Some(0),
total: true,
..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),
total: true,
..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> {
Expand Down
9 changes: 5 additions & 4 deletions modules/analysis/src/service/load/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
str::FromStr,
sync::Arc,
Expand Down Expand Up @@ -434,8 +434,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)
Expand All @@ -456,6 +456,7 @@ impl InnerService {
) -> Result<Vec<(Uuid, Arc<PackageGraph>)>, Error> {
let distinct_sbom_ids = sbom::Entity::find()
.filter(sbom::Column::SbomId.in_subquery(subquery))
.order_by_asc(sbom::Column::SbomId)
Comment on lines 457 to +459

@sourcery-ai sourcery-ai Bot Jul 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (performance): Ordering by SbomId at the database layer may have indexing implications

This adds deterministic ordering but may introduce an extra sort on the DB side if there’s no index on SbomId (or if it doesn’t match the PK/index). Please check whether an appropriate index exists or whether this ordering can reuse an existing one. If not, and the number of distinct_sbom_ids is usually small, consider sorting client-side instead.

Suggested implementation:

    ) -> Result<Vec<(Uuid, Arc<PackageGraph>)>, Error> {
        // Fetch distinct_sbom_ids without imposing a DB-side sort; we sort client-side below.
        let mut distinct_sbom_ids = sbom::Entity::find()
            .filter(sbom::Column::SbomId.in_subquery(subquery))
            .select()
            .all(connection)
            .await?;

        // Ensure deterministic ordering by sorting in memory by SbomId.
        distinct_sbom_ids.sort_by_key(|model| model.sbom_id);
  1. This change assumes the SeaORM model for the sbom table exposes the primary key/column as sbom_id. If the field name differs (e.g. sbom_id is nested or renamed), adjust the closure in sort_by_key accordingly.
  2. If distinct_sbom_ids is later used as an iterator or converted into another structure (e.g. mapping to (Uuid, Arc<PackageGraph>)), no other changes should be needed, since the vector is now deterministically ordered before subsequent processing.

.select()
.all(connection)
.await?
Expand Down
194 changes: 177 additions & 17 deletions modules/analysis/src/service/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -520,6 +520,176 @@ impl AnalysisService {
.await
}

/// Count nodes matching the query across all graphs, without hydrating them.
fn count_matching_nodes(query: &GraphQuery, graphs: &[(Uuid, Arc<PackageGraph>)]) -> u64 {
graphs
.iter()
.map(|(_, graph)| {
graph
.node_indices()
.filter(|&i| Self::filter(graph, query, i))
.count() as u64
})
.sum()
Comment on lines +524 to +533

@sourcery-ai sourcery-ai Bot Jul 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (performance): Avoid double iteration when computing totals for paginated queries to reduce cost on large graphs

When paginated.total() is true, this now walks all graphs twice: once in count_matching_nodes, and again in collect_graph_paged to build the page. On large graphs this doubles the dominant cost. Consider deriving the total from the same matching sequence used in collect_graph_paged (e.g., via a shared helper that returns both items and total, or by having collect_graph_paged optionally compute/return the total) so you only traverse and filter the nodes once per request.

Suggested implementation:

            })
    }

To fully adopt the "single traversal" approach and remove the double walk:

  1. Locate the function that builds paginated results (likely named something like collect_graph_paged or similar) in modules/analysis/src/service/mod.rs.
  2. Remove its dependency on count_matching_nodes:
    • Delete any calls to Self::count_matching_nodes(query, graphs) (or similar).
  3. In the body of collect_graph_paged, where you currently iterate over graphs/nodes to produce the items for the current page:
    • Introduce a total counter initialized to 0u64.
    • For each node that passes Self::filter(graph, query, i):
      • Increment total by 1.
      • Apply pagination (offset/limit) logic to decide whether to push that node into the items collection.
    • This way, you build the page and the total in the same loop.
  4. Update the return type or construction of the paginated response:
    • If the API allows, have collect_graph_paged return both items and total (or construct a struct that contains both) so callers can access the total without an extra traversal.
    • Only compute/attach total when paginated.total() (or equivalent flag) is true, to avoid unnecessary counting when totals are not requested.
  5. Ensure there are no remaining references to count_matching_nodes anywhere in the codebase; if there are, update them to use the new "single traversal" logic described above.

These changes will ensure that totals for paginated queries are derived during the same pass that builds the page, avoiding the double iteration and associated cost on large graphs.

}

/// 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<GraphQuery<'a>> + Debug,
graphs: &'g [(Uuid, Arc<PackageGraph>)],
concurrency: usize,
offset: usize,
limit: usize,
create: F,
) -> Result<Vec<Node>, Error>
where
F: Fn(&'g Graph<graph::Node, Relationship>, NodeIndex, &'g graph::Node) -> Fut + Clone,
Fut: Future<Output = Result<Node, Error>>,
{
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<_> = 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)
Comment on lines +575 to +580

@sourcery-ai sourcery-ai Bot Jul 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (bug_risk): Using buffer_unordered may break deterministic ordering of nodes within a page

buffer_unordered can change the order of results based on completion timing, so the final Vec may no longer be sorted by (sbom_id, node_id). If callers depend on stable pagination (consistent page contents or resuming by offset), this non-determinism could cause subtle bugs. Consider using buffered or re-sorting after collection to retain deterministic ordering while still leveraging concurrency.

.try_collect::<Vec<_>>()
.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<GraphQuery<'a>> + Debug,
options: QueryOptions,
graphs: &[(Uuid, Arc<PackageGraph>)],
paginated: impl Pagination,
connection: &C,
) -> Result<PaginatedResults<Node>, Error> {
let query = query.into();
let relationships = options.relationships;
tracing::debug!(?relationships, "relations");

let total = if paginated.total() {
Some(Self::count_matching_nodes(&query, graphs))
} else {
None
};

let limit = paginated.limit() as usize;
let offset = paginated.offset() as usize;

if limit == 0 {
return Ok(PaginatedResults {
items: vec![],
total,
});
}

if total.is_some_and(|t| offset as u64 >= t) {
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,
Expand Down Expand Up @@ -620,11 +790,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
Expand All @@ -641,11 +808,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(
Expand All @@ -667,19 +831,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());

let components = self
.run_graph_query(query, options, &graphs, connection)
.await?;
tracing::debug!("graph sbom count: {:?}", graphs.len());

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
Expand Down
Loading
Loading