Materialize ancestor information [Backport release/0.5.z]

[trustify-ci-bot]

Description

Backport of #2412 to release/0.5.z.

Summary by Sourcery

Materialize SBOM ancestor and describing CPE relationships to support faster and more reliable latest-SBOM analysis, refactor CPE resolution and graph loading to use these tables and bounded concurrency, and align APIs, tests, and dependencies with the new model.

New Features:

• Add materialized sbom_describing_cpe and sbom_ancestor tables and populate them during SBOM ingestion for Red Hat product-component relationships.
• Introduce configuration to limit latest-SBOM selection by published date cutoff and expose partial database migration support via a new migrate_up_to helper and CLI flag.
• Add utilities and tests for lazy connection pools, external graph traversal depth, SBOM lifecycle, and referential cleanup of materialized tables on SBOM deletion.

Bug Fixes:

• Fix ancestor resolution and ranking logic to avoid infinite recursion and to prefer checksum-verified ancestor matches, addressing node ID collision edge cases in Red Hat product-component graphs.
• Ensure external SBOM traversal consumes depth correctly so descendant limits are respected across SBOM boundaries.
• Prevent connection pool exhaustion by pinning deep descendant queries to a single read-only transaction and connection.

Enhancements:

• Refactor SBOM CPE resolution to batch queries, leverage materialized ancestor data, and de-duplicate results across phases before ranking.
• Simplify and parallelize graph loading with bounded concurrency and improved tracing, while removing unused recursive helper logic.
• Standardize endpoint error handling around a shared Error type and authorization errors, and introduce a reusable TruncatedIter formatter for concise logging output.

Build:

• Upgrade core dependencies including sea-orm, tokio, uuid, moka, async-compression, reqwest, and related workspace crates to newer versions.

Documentation:

• Document the Red Hat product-component model and the new ancestor materialization approach, and extend environment variable docs with the latest-SBOM published cutoff configuration.

Tests:

• Extend and add integration tests for CPE/materialization behavior, ancestry population, SBOM lifecycle across ingestion and deletion, external traversal depth, lazy pool behavior, and cleanup of describing CPE and ancestor links.
• Add focused tests around Red Hat node ID collisions, ranking semantics, and license/SBOM endpoint behavior with the new read-only transaction pattern.

Chores:

• Add container ignore configuration and new test data SBOM fixtures for Red Hat node ID collision and pool exhaustion scenarios.

[sourcery-ai]

Reviewer's Guide

Backport of the ancestor materialization and latest-SBOM ranking rework: CPE/context resolution is moved from recursive in-memory graph walks to materialized tables and batched SQL, graph loading is made concurrent and connection-safe, new migrations/entities for sbom_ancestor and sbom_describing_cpe are added, ingestion populates these tables, and APIs/tests/configs are updated accordingly.

Sequence diagram for latest SBOM CPE resolution using materialized ancestry

sequenceDiagram
    participant Client
    participant AnalysisService
    participant InnerService
    participant resolve_sbom_cpes
    participant batch_resolve_direct_cpe_matches
    participant batch_resolve_ancestor_cpes
    participant DB

    Client->>AnalysisService: retrieve_latest(...)
    AnalysisService->>InnerService: retrieve(...)
    InnerService->>DB: sbom_node::Entity::find() ... (matching sboms)
    DB-->>InnerService: Vec<Row>

    InnerService->>resolve_sbom_cpes: resolve_sbom_cpes(cpe_search, connection, rows)

    alt cpe_search == true
        resolve_sbom_cpes->>batch_resolve_direct_cpe_matches: batch_resolve_direct_cpe_matches(rows, connection)
        batch_resolve_direct_cpe_matches->>DB: query sbom_external_node + sbom_describing_cpe
        DB-->>batch_resolve_direct_cpe_matches: external_node_ref, cpe_ids
        batch_resolve_direct_cpe_matches->>DB: lookup nodes via sbom_node + sbom_ancestor
        DB-->>batch_resolve_direct_cpe_matches: candidate nodes
        batch_resolve_direct_cpe_matches-->>resolve_sbom_cpes: Vec<RankedSbom> (direct)
    end

    resolve_sbom_cpes->>batch_resolve_ancestor_cpes: batch_resolve_ancestor_cpes(sbom_ids, connection)
    batch_resolve_ancestor_cpes->>DB: query sbom_ancestor + sbom_describing_cpe (phase2a)
    DB-->>batch_resolve_ancestor_cpes: AncestorCpeRow (direct)
    batch_resolve_ancestor_cpes->>DB: recursive CTE on sbom_ancestor + sbom_describing_cpe (phase2b)
    DB-->>batch_resolve_ancestor_cpes: AncestorCpeRow (recursive)
    batch_resolve_ancestor_cpes-->>resolve_sbom_cpes: (phase2a, phase2b)

    resolve_sbom_cpes-->>InnerService: Vec<RankedSbom>
    InnerService->>InnerService: apply_rank(&mut ranked_sboms)
    InnerService->>InnerService: filter rank == 1

    InnerService->>InnerService: load_graphs(connection, latest_ids)
    InnerService->>DB: sbom_node / package_relates_to_package (per SBOM)
    DB-->>InnerService: graphs

    InnerService-->>AnalysisService: Vec<(Uuid, PackageGraph)>
    AnalysisService-->>Client: response with latest SBOMs

Loading

Entity relationship diagram for materialized SBOM ancestry and describing CPEs

erDiagram
    sbom {
        UUID sbom_id PK
    }

    sbom_describing_cpe {
        UUID sbom_id PK
        UUID cpe_id PK
    }

    sbom_ancestor {
        UUID sbom_id PK
        UUID ancestor_sbom_id PK
    }

    sbom_node_cpe_ref {
        UUID sbom_id
        UUID cpe_id
        string node_id
    }

    sbom_node_checksum {
        UUID sbom_id
        string node_id
        string value
    }

    sbom_external_node {
        UUID sbom_id
        string node_id
        string external_node_ref
        int external_type
    }

    sbom ||--o{ sbom_describing_cpe : has
    sbom ||--o{ sbom_ancestor : has_child
    sbom ||--o{ sbom_ancestor : has_ancestor

    sbom ||--o{ sbom_node_cpe_ref : has
    sbom ||--o{ sbom_node_checksum : has
    sbom ||--o{ sbom_external_node : has

    sbom_node_cpe_ref }o--|| sbom_describing_cpe : materialized_from
    sbom_node_checksum }o--o{ sbom_external_node : checksum_link
    sbom_node_checksum }o--|| sbom_ancestor : derives_child_ancestor

Loading

File-Level Changes

Change	Details	Files
Replace recursive ancestor resolution for latest SBOM ranking with batched SQL using materialized sbom_ancestor/sbom_describing_cpe tables and improved ranking logic.	Remove per-row recursive ancestor resolution (find_external_refs, find_node_ancestors, resolve_sbom_cpe) and introduce batch_resolve_direct_cpe_matches and batch_resolve_ancestor_cpes operating on all rows at once. Introduce AncestorCpeRow and ExternalNodeWithCpes models and use Postgres CTEs and array_agg to fetch direct and recursive ancestor CPEs in a few queries instead of N+1 graph walks. Refactor resolve_sbom_cpes to run in two phases (direct external CPEs, then ancestor CPEs), deduplicating phase-2b results against phase-1/2a to avoid duplicates while still allowing extra ancestry-derived components. Tighten apply_rank implementation (same semantics) over (cpe_id, matched_name, sbom_date) and switch Row/RankedSbom timestamps from DateTimeWithTimeZone to time::OffsetDateTime. Extend ResolvedSbom to carry cpe_ids (for RedHatProductComponent checksum-based resolution) and adjust resolve_rh_external_sbom_ancestors to aggregate CPEs per checksum match using sbom_describing_cpe.	modules/analysis/src/service/load/rank.rs modules/analysis/src/service/mod.rs modules/analysis/src/service/collector.rs modules/analysis/src/service/test/mod.rs modules/analysis/src/service/test/external_depth.rs
Materialize describing CPEs and ancestor relationships at ingest time via new tables/entities and migrations, and update advisory/product SQL to use them.	Add sbom_describing_cpe and sbom_ancestor entities and relations from sbom, sbom_external_node, and sbom_node_checksum, plus migrations to create/populate these tables and supporting indexes. Implement SbomContext::populate_describing_cpes and populate_ancestors and invoke them from both CycloneDX and SPDX ingestion paths to materialize describing CPEs and (child,ancestor) checksum-based links. Update raw SQL helpers (CONTEXT_CPE_FILTER_SQL, product_advisory_info_sql) to read describing CPEs from sbom_describing_cpe instead of recomputing describes-joins at query time and adjust existence checks accordingly. Add tests to verify sbom_describing_cpe and sbom_ancestor are populated on ingest, and cleaned up on SBOM delete (both child and ancestor directions) and across a full SBOM lifecycle with latest filters. Document the Red Hat product-component model and its checksum-based linking in docs/design/rh-product-component.md.	entity/src/sbom_describing_cpe.rs entity/src/sbom_ancestor.rs entity/src/sbom.rs entity/src/sbom_external_node.rs entity/src/sbom_node_checksum.rs migration/src/m0002110_sbom_describing_cpe.rs migration/src/m0002130_sbom_ancestor.rs migration/src/m0002120_ancestor_walk_index.rs modules/ingestor/src/graph/sbom/mod.rs modules/ingestor/src/graph/sbom/cyclonedx.rs modules/ingestor/src/graph/sbom/spdx.rs modules/fundamental/src/sbom/model/raw_sql.rs modules/fundamental/src/sbom/service/test.rs docs/design/rh-product-component.md
Rework graph loading and traversal to be concurrency-aware, connection-efficient, and depth-correct.	Change AnalysisService::load_graphs and InnerService::load_graphs to accept Vec, dedupe IDs, and load graphs concurrently using futures::stream::buffered(self.concurrency) instead of custom recursive load_graphs_inner and sbom_id helper. Add concurrency and sbom_published_cutoff_days to AnalysisConfig/InnerService; filter candidate SBOMs by published cutoff before ranking and improve logging using TruncatedIter. Adjust GraphLoader::load_graph to call the new load_graphs API and ensure Collector::with and forks decrement depth using saturating_sub so external SBOM hops consume depth; add external_depth test to validate behavior. Add LazyPool test context and deep_descendants_lazy_pool test to assert deep descendant queries use a single pool connection with read-only transactions and bounded concurrency.	modules/analysis/src/service/load/mod.rs modules/analysis/src/service/collector.rs modules/analysis/src/config.rs modules/analysis/src/service/test/external_depth.rs modules/analysis/src/endpoints/tests/mod.rs test-context/src/ctx/lazy_pool.rs test-context/src/lib.rs
Introduce helper formatting and DB utilities (truncated debug, read-only repeatable-read transactions, targeted migrations) and wire them through CLI and endpoints.	Add common::fmt::TruncatedIter for bounded debug/log output of large iterables and use it in analysis logging for SBOM IDs, ranked SBOMs, describing/unconnected nodes, and latest IDs. Introduce DatabaseExt::begin_read to start a REPEATABLE READ, READ ONLY transaction and switch multiple endpoints (sbom, purl, advisory, vulnerability, license, organization, product, analysis status/search) to return crate::Error and use db.begin() for consistent read-only use. Extend trustify_db::Database with migrate_up_to and helper find_migration_position to allow applying migrations up to a named migration (exact or substring match); update trustd db CLI Migrate command to accept --up-to and refactor Run methods to static fns taking Database. Add LazyPool::lazy_pool to test context to support small, lazily-created connection pools in tests.	common/src/fmt.rs common/src/db/mod.rs common/db/src/lib.rs trustd/src/db.rs modules/fundamental/src/sbom/endpoints/mod.rs modules/fundamental/src/sbom/endpoints/query.rs modules/fundamental/src/error.rs modules/fundamental/src/purl/endpoints/mod.rs modules/fundamental/src/purl/endpoints/base.rs modules/fundamental/src/advisory/endpoints/mod.rs modules/fundamental/src/advisory/endpoints/label.rs modules/fundamental/src/vulnerability/endpoints/mod.rs modules/fundamental/src/license/endpoints/mod.rs modules/fundamental/src/product/endpoints/mod.rs modules/fundamental/src/organization/endpoints/mod.rs modules/analysis/src/endpoints/mod.rs test-context/src/ctx/mod.rs
Update error handling, tests, and dependencies to support the new behavior and performance characteristics.	Extend analysis and fundamental Error enums to wrap AuthorizationError and ExternalReferenceQueryParseError and remove manual Display impl for the latter in favor of thiserror; normalize endpoint signatures to return Result<impl Responder, Error>. Add new tests for ancestor population, describing CPE cleanup, sbom lifecycle (ingest/upgrade/delete/re-ingest) with latest filters, and pool-exhaustion scenarios, including new SPDX/CycloneDX test data files. Bump several third-party crates (sea-orm, tokio, uuid, moka, async-compression features, reqwest features, storage uuid) and adjust imports (time macros instead of chrono, OffsetDateTime usage) accordingly. Update documentation to mention TRUSTIFY_LATEST_SBOM_PUBLISHED_CUTOFF_DAYS and add .containerignore for container builds.	modules/fundamental/src/sbom/endpoints/query.rs modules/fundamental/src/error.rs modules/analysis/src/error.rs modules/analysis/src/endpoints/tests/latest_filters.rs modules/fundamental/src/sbom/endpoints/mod.rs modules/analysis/src/config.rs docs/env-vars.md Cargo.toml common/Cargo.toml modules/analysis/Cargo.toml modules/storage/Cargo.toml test-context/Cargo.toml etc/test-data/** .containerignore

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey - I've found 1 issue, and left some high level feedback:

• The Describes relationship is hard-coded as 13 in multiple raw SQL snippets (e.g. populate_describing_cpes, CONTEXT_CPE_FILTER_SQL, product_advisory_info_sql); consider introducing a shared constant or helper to derive this from Relationship::Describes to avoid drift if the enum changes.
• In batch_resolve_ancestor_cpes the recursive CTE limits depth with a hard-coded < 10; it would be safer either to make this bound configurable or at least centralize it as a named constant with a comment explaining why 10 is appropriate for expected ancestor chain lengths.
• In resolve_sbom_cpes the dedup set uses HashSet<(Uuid, String)> and calls matched.name.clone() inside the loop for membership checks; you could avoid repeated allocations on this potentially hot path by keying the set on IDs only or using (&Uuid, &str) (e.g. via an interning map or by populating the set after building the final results).

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The `Describes` relationship is hard-coded as `13` in multiple raw SQL snippets (e.g. `populate_describing_cpes`, `CONTEXT_CPE_FILTER_SQL`, `product_advisory_info_sql`); consider introducing a shared constant or helper to derive this from `Relationship::Describes` to avoid drift if the enum changes.
- In `batch_resolve_ancestor_cpes` the recursive CTE limits depth with a hard-coded `< 10`; it would be safer either to make this bound configurable or at least centralize it as a named constant with a comment explaining why 10 is appropriate for expected ancestor chain lengths.
- In `resolve_sbom_cpes` the dedup set uses `HashSet<(Uuid, String)>` and calls `matched.name.clone()` inside the loop for membership checks; you could avoid repeated allocations on this potentially hot path by keying the set on IDs only or using `(&Uuid, &str)` (e.g. via an interning map or by populating the set after building the final `results`).

## Individual Comments

### Comment 1
<location path="modules/analysis/src/service/load/rank.rs" line_range="360-75" />
<code_context>
+
+    // Build dedup set from Phase 1 + Phase 2a using owned strings
+    // to avoid borrowing `results` while we push to it below
+    let existing: HashSet<(Uuid, String)> = results
+        .iter()
+        .map(|r| (r.cpe_id, r.matched_name.clone()))
+        .collect();
+
+    // Phase 2b results are deduped against the combined set — only
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Phase 2b dedup only considers pre-existing results, so duplicates within Phase 2b are still possible

`existing` is built once from `results` after Phases 1/2a and never updated as Phase 2b pushes new rows. If the recursive CTE yields the same `(cpe_id, matched_name)` multiple times (e.g., via different ancestor paths), all of them will be appended because only the first is checked against `existing`. If Phase 2b-internal duplicates are unwanted, consider updating `existing` after each insert or tracking a separate Phase 2b-local set.

Suggested implementation:

```rust
    // Build dedup set from Phase 1 + Phase 2a using owned strings
    // to avoid borrowing `results` while we push to it below. This set
    // is updated as Phase 2b inserts new rows so that Phase 2b-internal
    // duplicates are also filtered out.
    let mut existing: HashSet<(Uuid, String)> = results
        .iter()
        .map(|r| (r.cpe_id, r.matched_name.clone()))
        .collect();

```

```rust
    for ancestor_cpe in &phase2b {
        if let Some(matched_rows) = rows_by_sbom.get(&ancestor_cpe.sbom_id) {
            for matched in matched_rows {
                if existing.contains(&(ancestor_cpe.cpe_id, matched.name.clone())) {
                    continue;
                }

```

To fully implement Phase 2b-internal deduplication, you also need to update the Phase 2b insertion logic (in the same `for ancestor_cpe in &phase2b` loop):

1. After you push the new result into `results` (e.g., `results.push(...)` for Phase 2b), insert the corresponding `(cpe_id, matched_name)` into `existing`, for example:

```rust
let matched_name = matched.name.clone();

if existing.contains(&(ancestor_cpe.cpe_id, matched_name.clone())) {
    continue;
}

// Existing Phase 2b push; adapt field names to your struct
results.push(RankedMatch {
    cpe_id: ancestor_cpe.cpe_id,
    matched_name: matched_name.clone(),
    // ... other fields ...
});

// Update dedup set so later Phase 2b rows also see this entry
existing.insert((ancestor_cpe.cpe_id, matched_name));
```

2. Ensure `matched_name` in the insert matches the field used in the struct (`matched_name` or whatever the field is actually called). This guarantees that duplicates produced solely within Phase 2b are filtered out, not just duplicates against Phases 1/2a.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

suggestion (bug_risk): Phase 2b dedup only considers pre-existing results, so duplicates within Phase 2b are still possible

existing is built once from results after Phases 1/2a and never updated as Phase 2b pushes new rows. If the recursive CTE yields the same (cpe_id, matched_name) multiple times (e.g., via different ancestor paths), all of them will be appended because only the first is checked against existing. If Phase 2b-internal duplicates are unwanted, consider updating existing after each insert or tracking a separate Phase 2b-local set.

Suggested implementation:

    // Build dedup set from Phase 1 + Phase 2a using owned strings
    // to avoid borrowing `results` while we push to it below. This set
    // is updated as Phase 2b inserts new rows so that Phase 2b-internal
    // duplicates are also filtered out.
    let mut existing: HashSet<(Uuid, String)> = results
        .iter()
        .map(|r| (r.cpe_id, r.matched_name.clone()))
        .collect();

    for ancestor_cpe in &phase2b {
        if let Some(matched_rows) = rows_by_sbom.get(&ancestor_cpe.sbom_id) {
            for matched in matched_rows {
                if existing.contains(&(ancestor_cpe.cpe_id, matched.name.clone())) {
                    continue;
                }

To fully implement Phase 2b-internal deduplication, you also need to update the Phase 2b insertion logic (in the same for ancestor_cpe in &phase2b loop):

1. After you push the new result into results (e.g., results.push(...) for Phase 2b), insert the corresponding (cpe_id, matched_name) into existing, for example:

let matched_name = matched.name.clone();

if existing.contains(&(ancestor_cpe.cpe_id, matched_name.clone())) {
    continue;
}

// Existing Phase 2b push; adapt field names to your struct
results.push(RankedMatch {
    cpe_id: ancestor_cpe.cpe_id,
    matched_name: matched_name.clone(),
    // ... other fields ...
});

// Update dedup set so later Phase 2b rows also see this entry
existing.insert((ancestor_cpe.cpe_id, matched_name));

2. Ensure matched_name in the insert matches the field used in the struct (matched_name or whatever the field is actually called). This guarantees that duplicates produced solely within Phase 2b are filtered out, not just duplicates against Phases 1/2a.