## Summary The offline (air-gapped) malware advisory sync only consumes the PDS `full_dataset` (`/all`) snapshot. Online instances receive incremental `/delta` updates each cycle; offline instances do not — every refresh requires an admin to unpack a whole new `full_dataset`, which the connector then re-reads in full (resumable per shard, but not incremental). ## Current behaviour - `Gitlab::PackageMetadata::Connector::MalwareOffline#data_after` reads only `<base_uri>/v3/<purl_type>/full_dataset/` (a `checkpoint.json` with `{"until": <epoch>}` plus `<hexshard>.tar.zst` shards). It yields data only on a first sync or when a **newer** `full_dataset` snapshot arrives (`checkpoint.sequence < snapshot_until`), and then re-reads the whole snapshot. - `PackageMetadata::MalwareAdvisorySyncService#bulk_delta?` is gated to `storage_type == :pds`, so offline registries never take a delta path. ## Problem As the advisory corpus grows, air-gapped customers must repeatedly transfer and re-ingest the entire dataset for each refresh instead of applying small incremental updates — a large-file transfer + ingestion cost (related to the streaming work in #602885). There is no incremental channel for offline today. Malware is also the only offline data type without incremental updates. GLAD (advisory DB) and license offline sync go through the base `Gitlab::PackageMetadata::Connector::Offline`, which treats the bundle as an accumulating, sequenced file set (`<sequence>/<chunk>.ndjson`) and drops everything up to the checkpointed file, so a refresh ingests only the newly appended files. `MalwareOffline` overrides that with a single monolithic `full_dataset/` snapshot. This proposal brings malware offline in line with the other two. ## Proposal (iterable) Add an offline delta channel: - PDS exports a delta bundle for offline distribution (sequenced shards + checkpoint) alongside `full_dataset/`. - The offline connector applies it incrementally after the initial full snapshot, advancing checkpoints like the PDS delta path. - Sync routing lets offline registries use the delta path once past first sync. ## Implementation plan **Scope.** Contained in `Gitlab::PackageMetadata::Connector::MalwareOffline`. No change to `MalwareAdvisorySyncService`: offline registries already take the individual path (`bulk_delta?` is PDS-only), which calls `connector.data_after(checkpoint)` — and that method decides full vs. delta. So the whole change lives in one connector. **Vendor layout** under `<base_uri>/v3/<purl_type>/`: - `full_dataset/` — unchanged: `checkpoint.json` (`{"until": <epoch>}`) + `<hexshard>.tar.zst` shards (the baseline; mirrors PDS `/all`). - `delta/` — new: zero or more `<delta_epoch>.tar.zst` archives, each a zstd-tar of NDJSON chunks, named by its delta cursor (mirrors PDS `/delta`). Producing and packaging these for offline distribution is a PDS/CDot task, tracked separately. **Routing** (mirrors the PDS connector's `first_sync? -> /all, else -> /delta`): - First sync, or a newer `full_dataset` baseline dropped in, reads `full_dataset/` (existing logic, resumable per shard). Keeping the "newer baseline" precedence preserves today's re-snapshot workflow, so this is additive. - Otherwise, read `delta/` archives whose cursor is greater than `checkpoint.sequence`, oldest first. **Checkpointing.** Each delta archive's records carry `sequence = <delta cursor>` and `chunk = <per-entry chunk>`, so the sync's existing per-archive checkpoint advance moves `checkpoint.sequence` to each delta's cursor — identical to how PDS deltas advance. `first_sync?` flips to false once the baseline sets `sequence = until`, so later runs take the delta path on their own. An interrupted delta leaves the checkpoint at the previous cursor, so the whole archive is re-read next run (the upsert is idempotent). **Out of scope / follow-ups.** The PDS-side delta export and the offline packaging of the `delta/` bundle; the streaming memory fix (#602885), which applies to both paths; specs (to be added with the change). ## Proposed diff (no specs) ```diff diff --git a/ee/lib/gitlab/package_metadata/connector/malware_offline.rb b/ee/lib/gitlab/package_metadata/connector/malware_offline.rb index 34ef80b2a1f3..75ceb3e32e8d 100644 --- a/ee/lib/gitlab/package_metadata/connector/malware_offline.rb +++ b/ee/lib/gitlab/package_metadata/connector/malware_offline.rb @@ -5,22 +5,40 @@ module PackageMetadata module Connector # Offline connector for malware advisories (v3), for air-gapped instances. # - # The vendor path is self-sufficient: an admin unpacks the PDS `full_dataset` - # into `<base_uri>/v3/<purl_type>/full_dataset/`, which holds a - # checkpoint.json ({"until": <epoch>}) and one `<hexshard>.tar.zst` per shard - # (each a zstd-tar of a single NDJSON file). Unlike PDS, the shards are local - # compressed files, not signed URLs; they are read straight from disk. + # The vendor path is self-sufficient: an admin unpacks the PDS distribution + # under `<base_uri>/v3/<purl_type>/`: + # - full_dataset/ -- a checkpoint.json ({"until": <epoch>}) and one + # <hexshard>.tar.zst per shard (the baseline snapshot; mirrors GET /all). + # - delta/ -- zero or more <delta_epoch>.tar.zst archives, each a zstd-tar + # of NDJSON chunks named by their delta cursor (mirrors GET /delta), + # applied incrementally after the baseline. + # Unlike PDS the archives are local compressed files, not signed URLs; they + # are read straight from disk. # # Inherits Offline only for #file_prefix (base_uri + version_format / # registry_id); the file-walking is replaced here since the layout differs. class MalwareOffline < Offline FULL_DATASET_DIR = 'full_dataset' + DELTA_DIR = 'delta' CHECKPOINT_FILENAME = 'checkpoint.json' + # First sync (or a newer full_dataset baseline) reads the snapshot; + # afterwards the vendored deltas are applied incrementally. Mirrors the + # PDS connector's first_sync? -> /all, else -> /delta routing. + def data_after(checkpoint) + if checkpoint.first_sync? || newer_full_dataset?(checkpoint) + full_dataset_files(checkpoint) + else + delta_files(checkpoint) + end + end + + private + # Yields the vendor shards for the current snapshot, resuming from the last # ingested shard so an interrupted air-gapped sync does not re-read the # whole dataset. A fresh sync or a newer snapshot reads every shard. - def data_after(checkpoint) + def full_dataset_files(checkpoint) return [] unless File.exist?(checkpoint_path) snapshot_until = full_dataset_until @@ -33,25 +51,44 @@ def data_after(checkpoint) chunk = shard_file.delete_suffix('.tar.zst').to_i(16) next [] if resuming && chunk <= checkpoint.chunk.to_i - files_from(shard_file, snapshot_until, chunk) + files_from(shard_file, full_dataset_dir, snapshot_until, chunk) end end - private + # Yields the NDJSON records from each vendored delta newer than the + # checkpoint, oldest first. A delta archive is named by its unix-seconds + # cursor (matching PDS's `delta` timestamp) and holds flat NDJSON chunks; + # the sync advances the checkpoint to each delta's cursor as it ingests. + # An interrupted delta leaves the checkpoint at the previous cursor, so the + # whole archive is re-read next run (the upsert is idempotent). + def delta_files(checkpoint) + return [] unless File.directory?(delta_dir) + + cursor = checkpoint.sequence.to_i + Dir.glob('*.tar.zst', base: delta_dir) + .filter_map do |file| + sequence = delta_sequence(file) + [sequence, file] if sequence && sequence > cursor + end + .sort_by(&:first) + .lazy.flat_map { |sequence, file| files_from(file, delta_dir, sequence) } + end - # All shards share the snapshot `until`; the shard identifier (base-16) is - # the checkpoint chunk (each shard archive holds a single NDJSON file). + # Reads a local .tar.zst (a full-dataset shard or a delta) and wraps each + # NDJSON entry as a data file. `chunk` is forced for full-dataset shards + # (one file per shard); deltas fall back to the per-file chunk from the + # archive, exactly as the PDS connector does. # - # TODO: File.binread loads the whole compressed shard into memory before + # TODO: File.binread loads the whole compressed archive into memory before # TarZstReader decompresses it, also fully, into memory. Switch to zstd # streaming -- open the file and stream through the reader. # https://gitlab.com/gitlab-org/gitlab/-/work_items/602885 - def files_from(shard_file, snapshot_until, chunk) - bytes = File.binread(File.join(full_dataset_dir, shard_file)) + def files_from(archive, dir, sequence, chunk = nil) + bytes = File.binread(File.join(dir, archive)) reader = Connector::Archive::TarZstReader.new(bytes) reader.each_entry.map do |entry| - data_file_class.new(entry.io, snapshot_until, chunk) + data_file_class.new(entry.io, sequence, chunk || entry.chunk) end end @@ -59,6 +96,25 @@ def shard_files Dir.glob('*.tar.zst', base: full_dataset_dir).sort end + # A delta archive is named `<unix-seconds>.tar.zst`; returns that cursor, + # or nil for a name that is not a plain integer so it is skipped. + def delta_sequence(filename) + Integer(filename.delete_suffix('.tar.zst'), 10) + rescue ArgumentError, TypeError + nil + end + + # A fresh baseline -- a full_dataset whose `until` is newer than the + # checkpoint -- takes precedence over deltas, so an admin can re-baseline + # by dropping in a new snapshot; otherwise post-first-sync runs take the + # delta path. + def newer_full_dataset?(checkpoint) + return false unless File.exist?(checkpoint_path) + + snapshot_until = full_dataset_until + !snapshot_until.nil? && checkpoint.sequence.to_i < snapshot_until + end + def full_dataset_until parsed = ::Gitlab::Json::SafeParser.parse(File.read(checkpoint_path)) parsed['until'] if parsed.is_a?(Hash) @@ -70,6 +126,10 @@ def full_dataset_dir File.join(file_prefix, FULL_DATASET_DIR) end + def delta_dir + File.join(file_prefix, DELTA_DIR) + end + def checkpoint_path File.join(full_dataset_dir, CHECKPOINT_FILENAME) end ``` ## Related - Offline path: #594758 (air-gapped auth/download), #613635 (offline checkpoint bug), #602885 (stream shards rather than loading whole `.tar.zst` into memory). - Mirrors the online delta work: #607448 (bulk multi-registry `/delta`), #611958 (`use_all` catch-up). - Base offline connector that already gives GLAD/licenses incremental refreshes: `ee/lib/gitlab/package_metadata/connector/offline.rb`.