Recompute group prevented-package stats in project batches

Arpit Gogiarequested to merge
631890-group-stats-2-refresh-service into master

What does this MR do and why?

Part 2 of 7 for https://gitlab.com/gitlab-org/gitlab/-/work_items/631890.

This MR adds the service that actually computes and stores a group's precomputed prevented-package summary. DependencyFirewallPreventedPackage.window_flags(project_ids:, from_7:, from_30:) returns one row per card and package identifier for a batch of projects, with blocked and warned-only flags for both windows and the highest severity.

GroupRefreshService.new(group, batch_size: 500).execute walks the group's projects in batches of 500, reads each batch from a caught-up replica, or from the primary when replicas lag (the worker is :sticky), merges the package sets in the job, and upserts both summary rows (7 and 30 day windows) in one statement. Window dates are computed in UTC.

A parity spec checks that the stored counts equal today's query-time group counts for the same data, and that changing the batch size does not change the result.

RefreshGroupPreventedStatsWorker is a standard Sidekiq job that runs GroupRefreshService for one group. It deduplicates :until_executing, has a concurrency_limit of 5, retries 3 times, is deferred on database health signals, and re-checks the group's flag before running. Callers schedule it with RefreshGroupPreventedStatsWorker.schedule(group_ids).

Nothing schedules this worker yet; parts 3, 4 and 5 do.

For reviewers: the deduplication key uses the default 10-minute TTL, so if a long throttled backlog, for example after the daily run, outlasts it, a group can be enqueued twice, costing extra work but not wrong numbers.

How a refresh works 
RefreshGroupPreventedStatsWorker.perform(group_id)
 │  Group.find_by_id → skip unless dependency_firewall_group_prevented_stats is on for the root group
 │  (deduplicated until it starts executing, at most 5 at a time, 3 retries, deferred if the database is unhealthy)
 ▼
GroupRefreshService.new(group, batch_size: 500).execute
 │
 ├─ 1. Window starts, in UTC     7 days  → today − 6 at 00:00
 │                               30 days → today − 29 at 00:00
 │
 ├─ 2. Every project id under the group, subgroups included
 │      (NamespaceProjectIdsEachBatch)  →  [p1, p2, … pN]
 │
 ├─ 3. For each slice of 500 projects:
 │      DependencyFirewallPreventedPackage.window_flags(project_ids:, from_7:, from_30:)
 │        SQL: rows touched in the last 30 days, GROUP BY (card, purl)
 │        → per (card, purl): blocked_7, warned_only_7, max_severity_7,
 │                            blocked_30, warned_only_30, max_severity_30
 │
 ├─ 4. Merge the slices in Ruby, keyed by (card, purl)
 │        flags: OR        severities: MAX
 │
 ├─ 5. Count once per window: 5 card counts and 6 severity counts → 2 summary rows
 │
 └─ 6. upsert_all(rows, unique_by: [namespace_id, window_days])
         → dependency_firewall_namespace_prevented_stats
Data shape at each stage (example)

Example: a group with three projects (11, 12, 13), refreshed on 2026-10-16 with batch_size: 2.

 worker input
 ┌─────────────────────────────┐
 │ group_id : Integer          │   9970
 └──────────────┬──────────────┘
                ▼  Group.find_by_id, flag check on root
 1. window starts  Hash{ Integer => Time (UTC midnight) }
 ┌─────────────────────────────────────────────┐
 │ { 7  => 2026-10-10 00:00 UTC,               │
 │   30 => 2026-09-17 00:00 UTC }              │
 └──────────────┬──────────────────────────────┘
                ▼
 2. project ids  Array<Integer> → each_slice(batch_size)
 ┌─────────────────────────────────────────────┐
 │ [11, 12, 13]  →  [[11, 12], [13]]           │
 └──────────────┬──────────────────────────────┘
                ▼  one SQL statement per slice
 3. window_flags(project_ids:, from_7:, from_30:)
    in : ledger rows (project_id, rule_type, identifier, last_blocked_at, last_warned_at, severity)
    out: Array<Hash>, one per (card, purl) in the slice
 ┌──────────────────────────────────────────────────────────────────────────────┐
 │ slice 1: [{ card: 'vulnerabilities', identifier: 'pkg:npm/lodash@4.17.20',  │
 │             blocked_7: true,  warned_only_7: false, max_severity_7: 6,       │
 │             blocked_30: true, warned_only_30: false, max_severity_30: 7 },   │
 │           { card: 'malicious', identifier: 'pkg:npm/event-stream@3.3.6',     │
 │             blocked_7: false, warned_only_7: true,  max_severity_7: nil,     │
 │             blocked_30: false, warned_only_30: true, max_severity_30: nil }] │
 │ slice 2: [{ card: 'malicious', identifier: 'pkg:npm/event-stream@3.3.6',     │
 │             blocked_7: true,  warned_only_7: false, ... },                   │
 │           { card: 'license',   identifier: 'pkg:npm/left-pad@1.3.0',         │
 │             blocked_7: false, blocked_30: true, ... }]                       │
 └──────────────┬───────────────────────────────────────────────────────────────┘
                ▼  merge: flags OR, severities MAX (nil-safe)
 4. merged  Hash{ [card, purl] => Hash }  → .values → Array<Hash> (same row shape)
 ┌──────────────────────────────────────────────────────────────────────────────┐
 │ ['vulnerabilities', lodash]       => { blocked_7: T, max_sev_7: 6,           │
 │                                        blocked_30: T, max_sev_30: 7, ... }   │
 │ ['malicious',       event-stream] => { blocked_7: T, warned_only_7: T,       │
 │                                        blocked_30: T, warned_only_30: T }    │
 │ ['license',         left-pad]     => { blocked_7: F, blocked_30: T, ... }    │
 └──────────────┬───────────────────────────────────────────────────────────────┘
                ▼  stat_row(days, from, purls) for 7 and 30
 5. rows  Array<Hash>: 2 entries, one per window, keyed by table column
 ┌──────────────────────────────────────────────────────────────────────────────┐
 │ { namespace_id: 9970, window_days: 7,  window_start: 2026-10-10,             │
 │   vulnerabilities_prevented: 1, malicious_packages_prevented: 1,             │
 │   malicious_packages_warned_only: 1, license_risks_prevented: 0,             │
 │   license_risks_warned_only: 0,                                              │
 │   severity_critical: 0, severity_high: 1, severity_medium: 0, ... }          │
 │ { namespace_id: 9970, window_days: 30, window_start: 2026-09-17,             │
 │   vulnerabilities_prevented: 1, ..., license_risks_prevented: 1,             │
 │   severity_critical: 1, severity_high: 0, ... }                              │
 └──────────────┬───────────────────────────────────────────────────────────────┘
                ▼  upsert_all(rows, unique_by: [namespace_id, window_days])
 6. dependency_firewall_namespace_prevented_stats
 ┌──────────────────────────────────────────────────────────────────────────────┐
 │ namespace_id │ window_days │ window_start │ vulns │ mal │ mal_w │ lic │ crit │ high │
 │ 9970         │ 7           │ 2026-10-10   │ 1     │ 1   │ 1     │ 0   │ 0    │ 1    │
 │ 9970         │ 30          │ 2026-09-17   │ 1     │ 1   │ 1     │ 1   │ 1    │ 0    │
 └──────────────────────────────────────────────────────────────────────────────┘

Database

The window_flags batch query and the summary upsert are new. To list a group's projects, GroupRefreshService uses the existing Gitlab::Database::NamespaceProjectIdsEachBatch helper.

Measured with Joe on gitlab-production-main on 2026-10-16. GitLab.com's ledger had 41 rows, so the session clone was seeded the same way as the earlier dependency firewall plans: 300 rows for each of gitlab-org's 9,140 projects, plus 10 rows per project for a 0.5% sample, about 5.41 million rows, then ANALYZE. The summary table was created in the session, since MR 1 had not reached the snapshot yet. Batches use real gitlab-org project ids, 500 per batch.

Query Warm First run Plans
window_flags, one batch of 500 projects 182 ms median (152 to 200 ms) 0.3 to 7.0 s median warm batch, coldest batch
window_flags, 13 batches (6,100 projects) 2.2 s in total 26.5 s in total
window_flags, whole group in one statement (cross-check only) 5.0 s 8.3 s warm
NamespaceProjectIdsEachBatch for gitlab-org (4,167 statements, 1,613 subgroups) about 0.4 s of database time about 0.4 s namespace walk, project page
Summary upsert, 2 rows 2.95 ms (conflict update) 10.1 ms (insert) insert, conflict update

A full gitlab-org refresh (9,140 projects, 19 batches) comes to about 3.3 s of warm window_flags time, plus the helper's statements and one upsert. Every statement stays well under the 15 s statement timeout; the slowest was the first, cold batch at 7.0 s. A single statement over the whole group took 5.0 s warm, after a poor row estimate led to a sort-based GroupAggregate, so paging also keeps each statement short. Each batch is an index scan on i_dep_fw_prevented_packages_warned by project_id, reading about 150,000 buffers. The job then holds one merged entry per card and package, 60,000 for gitlab-org at this volume, about 18 MB.

Raw SQL and plans

window_flags, one batch (plan)

SELECT CASE rule_type WHEN 0 THEN 'license' WHEN 1 THEN 'vulnerabilities' WHEN 2 THEN 'malicious' WHEN 4 THEN 'vulnerabilities' WHEN 5 THEN 'package_version' END, "dependency_firewall_prevented_packages"."identifier" COLLATE "C", COALESCE(bool_or(last_blocked_at >= '2026-10-10 00:00:00'), FALSE) AS blocked_7, COALESCE(bool_or(last_warned_at >= '2026-10-10 00:00:00' AND (last_blocked_at IS NULL OR last_blocked_at < '2026-10-10 00:00:00')), FALSE) AS warned_only_7, MAX(severity) FILTER (WHERE last_blocked_at >= '2026-10-10 00:00:00') AS max_severity_7, COALESCE(bool_or(last_blocked_at >= '2026-09-17 00:00:00'), FALSE) AS blocked_30, COALESCE(bool_or(last_warned_at >= '2026-09-17 00:00:00' AND (last_blocked_at IS NULL OR last_blocked_at < '2026-09-17 00:00:00')), FALSE) AS warned_only_30, MAX(severity) FILTER (WHERE last_blocked_at >= '2026-09-17 00:00:00') AS max_severity_30 FROM "dependency_firewall_prevented_packages" WHERE "dependency_firewall_prevented_packages"."project_id" IN (<500 project ids>) AND ("dependency_firewall_prevented_packages"."last_blocked_at" >= '2026-09-17 00:00:00' OR "dependency_firewall_prevented_packages"."last_warned_at" >= '2026-09-17 00:00:00') GROUP BY CASE rule_type WHEN 0 THEN 'license' WHEN 1 THEN 'vulnerabilities' WHEN 2 THEN 'malicious' WHEN 4 THEN 'vulnerabilities' WHEN 5 THEN 'package_version' END, "dependency_firewall_prevented_packages"."identifier" COLLATE "C"
Plan, warm median batch 
 HashAggregate  (cost=47785.55..48189.52 rows=17954 width=72) (actual time=159.682..170.112 rows=43598 loops=1)
   Group Key: CASE dependency_firewall_prevented_packages.rule_type WHEN 0 THEN 'license'::text WHEN 1 THEN 'vulnerabilities'::text WHEN 2 THEN 'malicious'::text WHEN 4 THEN 'vulnerabilities'::text WHEN 5 THEN 'package_version'::text ELSE NULL::text END, (dependency_firewall_prevented_packages.identifier)::text
   Batches: 1  Memory Usage: 17937kB
   Buffers: shared hit=150889
   I/O Timings: read=0.000 write=0.000
   ->  Index Scan using i_dep_fw_prevented_packages_warned on public.dependency_firewall_prevented_packages  (cost=0.43..46973.83 rows=20293 width=82) (actual time=0.059..116.573 rows=83734 loops=1)
         Index Cond: (dependency_firewall_prevented_packages.project_id = ANY ('{<500 project ids>}'::bigint[]))
         Filter: ((dependency_firewall_prevented_packages.last_blocked_at >= '2026-09-17 00:00:00+00'::timestamp with time zone) OR (dependency_firewall_prevented_packages.last_warned_at >= '2026-09-17 00:00:00+00'::timestamp with time zone))
         Rows Removed by Filter: 65969
         Buffers: shared hit=150889
         I/O Timings: read=0.000 write=0.000
Settings: effective_cache_size = '472585MB', work_mem = '230MB', jit = 'off', random_page_cost = '1.5', seq_page_cost = '4'

Summary upsert (insert, conflict update)

INSERT INTO "dependency_firewall_namespace_prevented_stats" ("namespace_id","window_days","window_start","vulnerabilities_prevented","malicious_packages_prevented","malicious_packages_warned_only","license_risks_prevented","license_risks_warned_only","severity_critical","severity_high","severity_medium","severity_low","severity_info","severity_unknown","created_at","updated_at") VALUES (<group id>, 7, '2026-10-10', <counts...>, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), (<group id>, 30, '2026-09-17', <counts...>, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ON CONFLICT ("namespace_id","window_days") DO UPDATE SET updated_at=(CASE WHEN ("dependency_firewall_namespace_prevented_stats"."window_start" IS NOT DISTINCT FROM excluded."window_start" AND "dependency_firewall_namespace_prevented_stats"."vulnerabilities_prevented" IS NOT DISTINCT FROM excluded."vulnerabilities_prevented" AND "dependency_firewall_namespace_prevented_stats"."malicious_packages_prevented" IS NOT DISTINCT FROM excluded."malicious_packages_prevented" AND "dependency_firewall_namespace_prevented_stats"."malicious_packages_warned_only" IS NOT DISTINCT FROM excluded."malicious_packages_warned_only" AND "dependency_firewall_namespace_prevented_stats"."license_risks_prevented" IS NOT DISTINCT FROM excluded."license_risks_prevented" AND "dependency_firewall_namespace_prevented_stats"."license_risks_warned_only" IS NOT DISTINCT FROM excluded."license_risks_warned_only" AND "dependency_firewall_namespace_prevented_stats"."severity_critical" IS NOT DISTINCT FROM excluded."severity_critical" AND "dependency_firewall_namespace_prevented_stats"."severity_high" IS NOT DISTINCT FROM excluded."severity_high" AND "dependency_firewall_namespace_prevented_stats"."severity_medium" IS NOT DISTINCT FROM excluded."severity_medium" AND "dependency_firewall_namespace_prevented_stats"."severity_low" IS NOT DISTINCT FROM excluded."severity_low" AND "dependency_firewall_namespace_prevented_stats"."severity_info" IS NOT DISTINCT FROM excluded."severity_info" AND "dependency_firewall_namespace_prevented_stats"."severity_unknown" IS NOT DISTINCT FROM excluded."severity_unknown") THEN "dependency_firewall_namespace_prevented_stats".updated_at ELSE CURRENT_TIMESTAMP END),"window_start"=excluded."window_start","vulnerabilities_prevented"=excluded."vulnerabilities_prevented","malicious_packages_prevented"=excluded."malicious_packages_prevented","malicious_packages_warned_only"=excluded."malicious_packages_warned_only","license_risks_prevented"=excluded."license_risks_prevented","license_risks_warned_only"=excluded."license_risks_warned_only","severity_critical"=excluded."severity_critical","severity_high"=excluded."severity_high","severity_medium"=excluded."severity_medium","severity_low"=excluded."severity_low","severity_info"=excluded."severity_info","severity_unknown"=excluded."severity_unknown" RETURNING "id"
Plan, conflict update 
 Insert on public.dependency_firewall_namespace_prevented_stats  (cost=0.00..0.03 rows=2 width=82) (actual time=0.266..0.294 rows=2 loops=1)
   Buffers: shared hit=50
   WAL: records=8 fpi=0 bytes=642
   I/O Timings: read=0.000 write=0.000
   ->  Values Scan on "*VALUES*"  (cost=0.00..0.03 rows=2 width=82) (actual time=0.075..0.079 rows=2 loops=1)
         Buffers: shared hit=17
         I/O Timings: read=0.000 write=0.000
Trigger RI_ConstraintTrigger_c_4036802406 for constraint fk_rails_95b888f15d: time=2.596 calls=2
Settings: effective_cache_size = '472585MB', work_mem = '230MB', jit = 'off', random_page_cost = '1.5', seq_page_cost = '4'

NamespaceProjectIdsEachBatch (existing helper): one namespace walk statement per 100 subgroups (plan), then per subgroup a first-id query, an OFFSET 100 query and a range query (plan).

How to set up and validate locally

Run the specs:

bin/rspec ee/spec/services/security/dependency_firewall/prevented_stats/group_refresh_service_spec.rb ee/spec/models/security/dependency_firewall_prevented_package_spec.rb ee/spec/workers/security/dependency_firewall/refresh_group_prevented_stats_worker_spec.rb

To try it in bin/rails console (nothing schedules the worker yet, so run it inline):

  1. Pick a group with at least one project and enable the flag on its top-level group:

    group = Group.find_by_full_path('<group>')
project = group.all_projects.first
Feature.enable(:dependency_firewall_group_prevented_stats, group.root_ancestor)
worker = Security::DependencyFirewall::RefreshGroupPreventedStatsWorker.new
stats = -> { Security::DependencyFirewallNamespacePreventedStat.where(namespace_id: group.id).order(:window_days).map { |s| [s.window_days, s.window_start, s.to_counts] } }

  2. Record a baseline: worker.perform(group.id), then stats.call. It returns two rows, for the 7 and 30 day windows.

  3. Add three prevented packages to the project: a malicious package blocked now, a critical risk severity package blocked 10 days ago, and a license package warned but not blocked:

    model = Security::DependencyFirewallPreventedPackage
now = Time.current
model.bulk_upsert!([
  { project_id: project.id, rule_type: model.rule_types['malicious'], identifier: 'pkg:npm/df-test-malicious@1.0.0',
    first_seen_at: now, last_blocked_at: now, last_warned_at: nil, severity: nil },
  { project_id: project.id, rule_type: model.rule_types['risk_severity'], identifier: 'pkg:npm/df-test-vulnerable@1.0.0',
    first_seen_at: 10.days.ago, last_blocked_at: 10.days.ago, last_warned_at: nil, severity: model.severities['critical'] },
  { project_id: project.id, rule_type: model.rule_types['license'], identifier: 'pkg:npm/df-test-license@1.0.0',
    first_seen_at: now, last_blocked_at: nil, last_warned_at: now, severity: nil }
])

  4. Refresh again with worker.perform(group.id) and compare stats.call with the baseline:

    • 7 day row: malicious prevented +1 and license warned only +1. Vulnerabilities are unchanged, since that block is 10 days old.
    • 30 day row: vulnerabilities prevented +1 (critical +1), malicious prevented +1 and license warned only +1.

  5. The batch size does not change the result: Security::DependencyFirewall::PreventedStats::GroupRefreshService.new(group, batch_size: 1).execute, then stats.call returns the same rows.

  6. With the flag off, the worker writes nothing:

    Feature.disable(:dependency_firewall_group_prevented_stats, group.root_ancestor)
Security::DependencyFirewallNamespacePreventedStat.where(namespace_id: group.id).delete_all
worker.perform(group.id)
stats.call # => []

  7. Clean up: model.where(identifier: %w[pkg:npm/df-test-malicious@1.0.0 pkg:npm/df-test-vulnerable@1.0.0 pkg:npm/df-test-license@1.0.0]).delete_all.

MR acceptance checklist

Evaluate this MR against the MR acceptance checklist. It helps you analyze changes to reduce risks in quality, performance, reliability, security, and maintainability.

References

Edited by Arpit Gogia

Merge request reports