Add malware advisory sync worker, configuration and scheduling

Bala Kumarrequested to merge
602432-sync-worker-config into master

What does this MR do and why?

Wire the malware advisory sync (epic gitlab-org&20876) into a scheduled, gated job — the Wave-2 orchestration slice. Extracted from the POC !226660 (closed).

Changes

  • MalwareAdvisorySyncService — per-PURL iteration over the configs from the shared PackageMetadata::SyncConfiguration (which already builds the malware_advisories configs from the existing package_metadata_purl_types setting and selects the offline-vs-PDS connector). Runs under a lease-backed StopSignal.
    • Per-archive (atomic) checkpoint advance: all chunks of one archive share a sequence, and the checkpoint only advances once an archive is fully ingested. An interrupt mid-archive leaves the checkpoint at the last complete archive, so the next run re-fetches the incomplete one rather than skipping its un-ingested chunks.
    • Lifecycle logging (started / completed / interrupted / skipped) correlated by batch_id (one worker run) + sync_id (one PURL sync).
  • MalwareAdvisoriesSyncWorker + cron (ee/config/schedule.yml, every 5 min), 22-min lease.
  • sync_malware_advisories feature flag (type: wip, default off).

Gating — the scheduled job is an inert no-op until both are in place

should_run? returns false unless both of:

  1. the dependency_scanning license feature is available,
  2. the sync_malware_advisories feature flag is enabled (default off).

Sizing

Longer window than the shared sync (22-min lease / 20-min MAX_SYNC_DURATION, vs 6/4) so a full first-sync bootstrap of up to ~30 MB (~365k advisories) fits in one run. Larger files are deferred to streaming decompression (#602885) and a resumable per-shard first sync (#603628).

Dependency & merge safety

This service names the ingestion classes from #602431 (!240168 (merged)) — MalwareAdvisoryIngestionService / MalwareAdvisoryDataObjectFabricator — but only inside method bodies, so the file loads on master without them and the specs stub them. With the feature flag off, nothing executes until both that MR lands and the flag is enabled, so this MR is safe to merge independently of #602431.

Local testing

Drove MalwareAdvisorySyncService end-to-end on a GDK against a fake connector yielding two GLAM v3 NDJSON archives (the connectors themselves are #602430, already merged), with the real ingestion pipeline present:

worker.should_run? (FF off) => false           # cron is a no-op by default
advisories ingested: +2
checkpoint after sync: sequence=2000 chunk=0    # advanced to the last complete archive

Confirms the worker is a no-op with the flag off, the service ingests via the real pipeline, and the checkpoint advances per archive.

Specs

bundle exec rspec \
  ee/spec/services/package_metadata/malware_advisory_sync_service_spec.rb \
  ee/spec/workers/package_metadata/malware_advisories_sync_worker_spec.rb
# 14 examples, 0 failures

MR acceptance checklist

  • Scheduled worker runs only when the license + flag are both present; a no-op otherwise.
  • Checkpoint advances atomically per archive; an interrupt re-runs the incomplete archive.
  • Specs for the service (checkpoint / interrupt / logging) and the worker (gating).
  • Feature-flag gated (sync_malware_advisories, default off); safe to merge before the ingestion MR lands.

Related to #602432.

Database review — upsert query plans

The sync drives ingestion in slices of INGEST_SLICE_SIZE = 1000, and bulk_upsert! re-batches to BulkInsertSafe::DEFAULT_BATCH_SIZE = 500, so the database sees 500-row INSERT … ON CONFLICT … DO UPDATE upserts — two per slice, one per malware table (both on the sec database). Statements captured from a local ingestion run of a 500-record batch (advisory statement ≈187 KB, affected ≈52 KB — abbreviated to their shape below).

1. pm_malware_advisories — 500-row upsert

INSERT INTO "pm_malware_advisories"
  ("advisory_xid", "source_xid", "title", "description", "published_date",
   "withdrawn_date", "identifiers", "urls", "created_at", "updated_at")
VALUES (...), (...)   -- 500 value rows
ON CONFLICT ("advisory_xid", "source_xid") DO UPDATE SET
  "title" = excluded."title", "description" = excluded."description",
  "published_date" = excluded."published_date", "withdrawn_date" = excluded."withdrawn_date",
  "identifiers" = excluded."identifiers", "urls" = excluded."urls",
  "updated_at" = excluded."updated_at"
RETURNING "advisory_xid", "id";

Plans (Database Lab — gitlab-production-sec clone, 500-row batch):

Conflict arbiter index i_pm_malware_advisories_xid_source — no sequential scan.

2. pm_malware_affected_packages — 500-row upsert

INSERT INTO "pm_malware_affected_packages"
  ("pm_malware_advisory_id", "purl_type", "package_name", "affected_range", "created_at", "updated_at")
VALUES (...), (...)   -- 500 value rows
ON CONFLICT ("pm_malware_advisory_id", "purl_type", "package_name") DO UPDATE SET
  "affected_range" = excluded."affected_range", "updated_at" = excluded."updated_at";

Plans (Database Lab — gitlab-production-sec clone, 500-row batch):

Conflict arbiter index i_pm_malware_affected_packages_unique_for_upsert — no sequential scan. Because pm_malware_advisory_id is an FK, the Database Lab batch sourced its 500 ids from existing advisories (SELECT id FROM pm_malware_advisories ORDER BY random() LIMIT 500), so the linked plans include a small sample/scan node ahead of the upsert; the upsert/conflict cost is the relevant part.

Database Lab

Both upserts resolve conflicts through their unique indexes (i_pm_malware_advisories_xid_source, i_pm_malware_affected_packages_unique_for_upsert) — no sequential scans. Measured at the real 500-row batch size on the gitlab-production-sec clone:

  • pm_malware_advisories — INSERT and UPDATE-on-conflict plans linked above. ✅
  • pm_malware_affected_packages — INSERT and UPDATE-on-conflict plans linked above. ✅
Edited by Bala Kumar

Merge request reports