diff --git a/backend/src/osspckgs/migrations/V1784567585__advisory_affected_ranges_soft_delete.sql b/backend/src/osspckgs/migrations/V1784567585__advisory_affected_ranges_soft_delete.sql new file mode 100644 index 0000000000..95258c5fd2 --- /dev/null +++ b/backend/src/osspckgs/migrations/V1784567585__advisory_affected_ranges_soft_delete.sql @@ -0,0 +1,5 @@ +ALTER TABLE advisory_affected_ranges ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ; + +CREATE INDEX IF NOT EXISTS advisory_affected_ranges_live_idx + ON advisory_affected_ranges (advisory_package_id) + WHERE deleted_at IS NULL; diff --git a/docs/adr/0001-oss-packages-design-decisions.md b/docs/adr/0001-oss-packages-design-decisions.md index f44602d32d..7d9021f7f9 100644 --- a/docs/adr/0001-oss-packages-design-decisions.md +++ b/docs/adr/0001-oss-packages-design-decisions.md @@ -223,7 +223,7 @@ Five sub-workers run concurrently (npm, Maven, OSV, GitHub, Docker Hub), all wri | `versions` | Append-only via `INSERT … ON CONFLICT DO NOTHING`. Yanked/deprecated status is a separate targeted `UPDATE (is_yanked = true) WHERE …`. | | `repos` | Registry workers (npm, Maven) do **not** write `repos` enrichment metadata. They INSERT a minimal `repos(url, host)` row — `url` (canonical) and `host` (coarse classification) are both derived from the declared repository URL — solely to create the FK target their `package_repos` link needs. `owner`/`name`/`stars`/`description` and all other metadata stay NULL and remain enricher-owned; existing rows are never updated by registry workers. The GitHub enricher — triggered when `repos.last_synced_at IS NULL` — upserts `repos` with metadata. Docker Hub worker adds `docker_*` columns on top. | | `package_repos` | Composite PK `(package_id, repo_url)`. Each `source` value ('declared', 'deps_dev', 'heuristic', 'manual') is a separate row — sources do not overwrite each other. | -| `advisories` | Upsert on `osv_id`. OSV is the single source of truth; no other worker writes to this table. | +| `advisories` | Upsert on `osv_id`. OSV is the source of truth for ongoing updates; deps.dev also writes this table at backfill / new-package time (see §Source of truth: deps.dev backfill vs registries / OSV) — OSV's writes are expected to overwrite deps.dev's over time, not the other way around. | | `maintainers` / `package_maintainers` | `maintainers`: upsert on `(ecosystem, username)`, never deleted — the identity history is preserved. `package_maintainers`: reflects the **current** link set — the npm worker replaces a package's links each ingest (delete + reinsert), so prior link rows are not retained. | | `downloads_daily` | Append-only time-series. Each `(package_id, date)` row is written once. npm and Maven workers own disjoint rows by ecosystem. Historical timelines are preserved — workers do not overwrite past dates. | | `downloads_last_30d` | Upsert on `(purl, end_date)`. Written by the weekly ranking worker only. The cached `packages.downloads_last_30d` column must be updated in the same pass. | @@ -364,6 +364,27 @@ Local verification against the live OSV dataset (2026-05-28) showed the multi-ra --- +### `advisory_affected_ranges` delete/dedup strategy + +Two duplication problems surfaced in Tinybird for `advisory_affected_ranges` (CM-1258): + +**Problem A — zombie rows from hard-delete + reinsert.** The OSV worker's per-sync flow (`upsertAdvisory.ts` → `deleteOsvOnlyRanges` in `services/libs/data-access-layer/src/packages/osv.ts`) hard-deletes all OSV-owned ranges for an advisory_package and reinserts the deduped set via plain `INSERT` (no `ON CONFLICT`), so an unchanged range gets a brand-new PK every sync. `advisory_affected_ranges` is Sequin-replicated with `REPLICA IDENTITY FULL` into a Tinybird `ReplacingMergeTree(updatedAt)` datasource that has no sign/`is_deleted` column and no delete-shaped ingest path — the DELETE never propagates downstream, only the new-PK reinsert does. Net effect: every sync produces one zombie row per range, even when nothing changed. + +**Problem B — deps.dev raw rows and OSV structured rows never collide.** deps.dev writes `range_raw`/`unaffected_raw`; OSV writes `introduced_version`/`fixed_version`/`last_affected`. Per §`advisory_affected_ranges` uniqueness scope above, the unique index is on the structured columns only, so a deps.dev row (structured columns NULL) never conflicts with the equivalent OSV row — both survive as separate rows for the same real-world range, even though §Source of truth already says OSV should overwrite deps.dev's data over time. + +**Decided fix — diff-based upsert with soft-delete, not sweep-and-reinsert:** + +- Add a `deleted_at` column to `advisory_affected_ranges` (soft-delete, same pattern as `V1783036800__security_contacts_soft_delete.sql`). +- Per advisory_package, per sync: read existing live OSV-owned tuples, diff against the new deduped tuple set. Tuple in both → skip (no write, no event). Tuple only in new → `INSERT ... ON CONFLICT (advisory_package_id, COALESCE(introduced_version,''), COALESCE(fixed_version,''), COALESCE(last_affected,'')) DO UPDATE SET updated_at = NOW(), deleted_at = NULL`. Tuple only in old → soft-delete (`SET deleted_at = NOW()`), not hard-delete. An unchanged advisory_package now produces zero writes and zero Sequin/Tinybird events per sync. +- Soft-delete (not hard-delete) is required because the unique key is the value tuple itself: when OSV corrects a range (e.g. narrows `fixed_version`), the corrected tuple is a different key, so the old tuple's row has no successor row to collapse into — without a `deleted_at` flag it would sit forever as a false-positive vulnerable-range match. +- After OSV writes its structured ranges for an advisory_package, soft-delete any live deps.dev-owned row (`range_raw IS NOT NULL OR unaffected_raw IS NOT NULL`) for that same advisory_package — this is the §Source of truth overwrite rule, made to actually take effect now that raw and structured rows are reconciled instead of coexisting. +- Tinybird: add `deletedAt` to `advisoryAffectedRanges.datasource` and filter `deletedAt IS NULL` in `ossPackages_enriched.pipe`'s join, alongside the existing `FINAL` modifier (`FINAL` only collapses rows sharing a sorting key — it does not merge two distinct PKs claiming the same real-world range, which is why this needs an explicit filter rather than relying on `FINAL` alone). +- **Rollout note**: the `deletedAt IS NULL` filter only stops *new* duplicates. Zombie rows already replicated into Tinybird under the old hard-delete + reinsert path have no surviving Postgres row to emit a `deleted_at` update from, so they keep reading as `deletedAt IS NULL` forever. Deploying this fix must be paired with a one-time truncate + full Sequin resnapshot of the `advisoryAffectedRanges` datasource from the live Postgres table, or the existing false-positive vulnerable ranges persist. + +**Decided**: 2026-07-20 + +--- + ### Per-source ingestion strategies #### npm @@ -576,6 +597,7 @@ The writer of a `downloads_last_30d` row must also update the cached `packages.d - 2026-05-29 — clarified `packages_universe` import semantics (one-time backfill + weekly snapshot-diff incrementals; the ranking job updates score/flag columns in place). Added §Source of truth: deps.dev backfill vs registries / OSV with lifecycle ownership rules and the agreed `package_source_log` provenance table (`(package_id, source)` PK; `columns` array tracks `table.column` paths each source writes). - 2026-05-29 — added §Criticality scoring methodology (graph signals — transitive dependent count and PageRank centrality; per-ecosystem percentile-rank formula in `[0, 1]`; floor + ceiling tier budget policy; `package_criticality_spotlight` table). - 2026-06-08 — retired `packages_universe` table. Signals (`downloads_last_30d`, `centrality_score`, `rank_in_ecosystem`) migrated onto `packages`. Both `rank_packages_universe()` overloads dropped; replaced by `rank_packages()` operating directly on `packages`, scoped to ecosystems present in `critical_top_n_by_ecosystem` JSONB. §Downloads timeline by tier simplified (Tier 2/3 split removed). §Write semantics `packages_universe` row removed. +- 2026-07-20 (CM-1258) — corrected stale §Write semantics `advisories` row (deps.dev also writes this table; OSV overwrites over time per §Source of truth, it isn't the sole writer). Added §`advisory_affected_ranges` delete/dedup strategy: diff-based upsert + soft-delete (`deleted_at`) replaces hard-delete + reinsert to stop zombie rows in Tinybird, and OSV now supersedes deps.dev's raw rows on overlap so the two sources stop coexisting as duplicates. --- ## Note on promotion to production diff --git a/services/apps/packages_worker/src/deps-dev/workflows/ingestAdvisories.ts b/services/apps/packages_worker/src/deps-dev/workflows/ingestAdvisories.ts index 543e79b118..96927a24d5 100644 --- a/services/apps/packages_worker/src/deps-dev/workflows/ingestAdvisories.ts +++ b/services/apps/packages_worker/src/deps-dev/workflows/ingestAdvisories.ts @@ -77,7 +77,19 @@ LEFT JOIN packages p ON p.purl = s.purl ON CONFLICT (advisory_id, ecosystem, package_name) DO NOTHING ` -// Separate statement — must execute after ADVISORY_PACKAGES_MERGE_SQL so advisory_packages rows exist +// Separate statement — must execute after ADVISORY_PACKAGES_MERGE_SQL so advisory_packages rows exist. +// Skips any advisory_package that already has a live OSV-owned range: OSV is the +// source of truth over deps.dev for overlapping advisory_packages (ADR-0001 +// §advisory_affected_ranges delete/dedup strategy), and this merge runs on its own +// BQ-driven schedule, independent of and potentially after an OSV sync — the +// per-tuple ON CONFLICT below can't see that, since a NULL-bounds raw tuple has a +// different key than OSV's structured tuple and would insert as a new live +// duplicate. The NOT EXISTS guard makes the ownership rule package-level instead +// of tuple-level, so deps.dev never adds a live row once OSV owns the package. +// ON CONFLICT revives (not skips) a soft-deleted row occupying the same tuple — +// typical after supersedeDepsDevRanges soft-deletes deps.dev rows on OSV takeover +// and OSV later drops the package again — otherwise DO NOTHING would leave +// staging's live data with no corresponding live row. const ADVISORY_AFFECTED_RANGES_MERGE_SQL = ` INSERT INTO advisory_affected_ranges (advisory_package_id, range_raw, unaffected_raw, introduced_version, created_at, updated_at) SELECT @@ -91,7 +103,38 @@ JOIN advisories adv ON adv.osv_id = s.osv_id JOIN advisory_packages ap ON ap.advisory_id = adv.id AND ap.ecosystem = s.ecosystem AND ap.package_name = s.package_name -ON CONFLICT (advisory_package_id, COALESCE(introduced_version, ''), COALESCE(fixed_version, ''), COALESCE(last_affected, '')) DO NOTHING +WHERE NOT EXISTS ( + SELECT 1 FROM advisory_affected_ranges live + WHERE live.advisory_package_id = ap.id + AND live.deleted_at IS NULL + AND live.range_raw IS NULL + AND live.unaffected_raw IS NULL +) +ON CONFLICT (advisory_package_id, COALESCE(introduced_version, ''), COALESCE(fixed_version, ''), COALESCE(last_affected, '')) +DO UPDATE SET + updated_at = NOW(), + deleted_at = NULL, + range_raw = EXCLUDED.range_raw, + unaffected_raw = EXCLUDED.unaffected_raw +WHERE advisory_affected_ranges.deleted_at IS NOT NULL +` + +// Runs as prepareSql (uncounted) ahead of ADVISORY_AFFECTED_RANGES_MERGE_SQL, in the +// same transaction. Row-locks every advisory_packages this chunk will touch, in id +// order, before the NOT EXISTS ownership check runs. OSV's upsertOne (services/apps/ +// packages_worker/src/osv/upsertAdvisory.ts) takes the matching per-row lock before it +// writes advisory_affected_ranges for that advisory_package, so whichever transaction +// (deps.dev chunk or OSV record) locks the row first now forces the other to wait and +// see its committed writes — closing the race the two independently-scheduled write +// paths would otherwise have on the ownership check (ADR-0001 §Write semantics). +const ADVISORY_PACKAGES_LOCK_SQL = ` +SELECT ap.id +FROM advisory_packages ap +JOIN staging.osspckgs_advisory_packages_raw s + ON s.ecosystem = ap.ecosystem AND s.package_name = ap.package_name +JOIN advisories adv ON adv.osv_id = s.osv_id AND adv.id = ap.advisory_id +ORDER BY ap.id +FOR UPDATE OF ap ` const ADVISORIES_PG_COLUMNS = [ @@ -259,6 +302,7 @@ export async function ingestAdvisories(opts: { const { rowsAffected, tableRowCounts } = await mergeStagingToTable({ jobId: pkgsExport.jobId, + prepareSql: ADVISORY_PACKAGES_LOCK_SQL, mergeSql: [ADVISORY_PACKAGES_MERGE_SQL, ADVISORY_AFFECTED_RANGES_MERGE_SQL], tableNames: ['advisory_packages', 'advisory_affected_ranges'], isFinal, diff --git a/services/apps/packages_worker/src/osv/__tests__/reconcileOsvRanges.integration.test.ts b/services/apps/packages_worker/src/osv/__tests__/reconcileOsvRanges.integration.test.ts new file mode 100644 index 0000000000..b25ad13157 --- /dev/null +++ b/services/apps/packages_worker/src/osv/__tests__/reconcileOsvRanges.integration.test.ts @@ -0,0 +1,220 @@ +import { afterAll, beforeAll, describe, expect, it } from 'vitest' + +import { + reconcileOsvRanges, + supersedeDepsDevRanges, +} from '@crowd/data-access-layer/src/packages/osv' +import { QueryExecutor, pgpQx } from '@crowd/data-access-layer/src/queryExecutor' +import { getDbConnection } from '@crowd/database' + +// Integration test: hits the running packages-db. Skipped automatically when +// any of the DB env vars are missing, matching deriveCriticalFlag.integration.test.ts. +const HAVE_DB = + !!process.env.CROWD_PACKAGES_DB_WRITE_HOST && + !!process.env.CROWD_PACKAGES_DB_PORT && + !!process.env.CROWD_PACKAGES_DB_USERNAME && + !!process.env.CROWD_PACKAGES_DB_DATABASE && + !!process.env.CROWD_PACKAGES_DB_PASSWORD + +const FIXTURE_OSV_ID = 'osv-test-fixture-reconcile' + +interface RangeRow { + introduced_version: string | null + fixed_version: string | null + last_affected: string | null + range_raw: string | null + unaffected_raw: string | null + deleted_at: string | null + updated_at: string +} + +async function cleanupFixture(qx: QueryExecutor): Promise { + await qx.result( + ` + DELETE FROM advisory_affected_ranges + WHERE advisory_package_id IN ( + SELECT ap.id FROM advisory_packages ap + JOIN advisories a ON a.id = ap.advisory_id + WHERE a.osv_id = $(osvId) + ) + `, + { osvId: FIXTURE_OSV_ID }, + ) + await qx.result( + `DELETE FROM advisory_packages WHERE advisory_id IN (SELECT id FROM advisories WHERE osv_id = $(osvId))`, + { osvId: FIXTURE_OSV_ID }, + ) + await qx.result(`DELETE FROM advisories WHERE osv_id = $(osvId)`, { osvId: FIXTURE_OSV_ID }) +} + +async function liveRanges(qx: QueryExecutor, advisoryPackageId: number): Promise { + return qx.select( + ` + SELECT introduced_version, fixed_version, last_affected, range_raw, unaffected_raw, deleted_at, updated_at + FROM advisory_affected_ranges + WHERE advisory_package_id = $(advisoryPackageId) + ORDER BY id + `, + { advisoryPackageId }, + ) +} + +describe.skipIf(!HAVE_DB)('reconcileOsvRanges / supersedeDepsDevRanges — real packages-db', () => { + let qx: QueryExecutor + let advisoryPackageId: number + + beforeAll(async () => { + const conn = await getDbConnection({ + host: process.env.CROWD_PACKAGES_DB_WRITE_HOST ?? '', + port: parseInt(process.env.CROWD_PACKAGES_DB_PORT ?? '0', 10), + database: process.env.CROWD_PACKAGES_DB_DATABASE ?? '', + user: process.env.CROWD_PACKAGES_DB_USERNAME ?? '', + password: process.env.CROWD_PACKAGES_DB_PASSWORD ?? '', + }) + qx = pgpQx(conn) + await cleanupFixture(qx) + + const advisory = await qx.selectOne( + ` + INSERT INTO advisories (osv_id, source, source_url, aliases, severity, cvss, cvss_source, created_at, updated_at) + VALUES ($(osvId), 'GHSA', NULL, ARRAY[]::text[], 'HIGH', 7.5, 'osv_cvss_v3', NOW(), NOW()) + RETURNING id + `, + { osvId: FIXTURE_OSV_ID }, + ) + const advisoryPackage = await qx.selectOne( + ` + INSERT INTO advisory_packages (advisory_id, package_id, ecosystem, package_name, created_at, updated_at) + VALUES ($(advisoryId), NULL, 'npm', 'reconcile-fixture-pkg', NOW(), NOW()) + RETURNING id + `, + { advisoryId: advisory.id }, + ) + advisoryPackageId = advisoryPackage.id as number + }, 30_000) + + afterAll(async () => { + if (qx) await cleanupFixture(qx) + }) + + it('inserts the initial OSV range set as live rows', async () => { + await reconcileOsvRanges(qx, advisoryPackageId, [ + { + advisoryPackageId, + introducedVersion: '1.0.0', + fixedVersion: '1.2.0', + lastAffected: null, + }, + { + advisoryPackageId, + introducedVersion: '2.0.0', + fixedVersion: null, + lastAffected: '2.5.0', + }, + ]) + + const rows = await liveRanges(qx, advisoryPackageId) + expect(rows).toHaveLength(2) + expect(rows.every((r) => r.deleted_at === null)).toBe(true) + }) + + it('leaves an unchanged tuple untouched on a no-op resync (no updated_at bump)', async () => { + const before = await liveRanges(qx, advisoryPackageId) + const beforeUpdatedAt = before.find((r) => r.introduced_version === '1.0.0')?.updated_at + + await new Promise((resolve) => setTimeout(resolve, 1100)) + await reconcileOsvRanges(qx, advisoryPackageId, [ + { + advisoryPackageId, + introducedVersion: '1.0.0', + fixedVersion: '1.2.0', + lastAffected: null, + }, + { + advisoryPackageId, + introducedVersion: '2.0.0', + fixedVersion: null, + lastAffected: '2.5.0', + }, + ]) + + const after = await liveRanges(qx, advisoryPackageId) + const afterUpdatedAt = after.find((r) => r.introduced_version === '1.0.0')?.updated_at + expect(afterUpdatedAt).toBe(beforeUpdatedAt) + }) + + it('soft-deletes a stale tuple dropped from the new range set', async () => { + await reconcileOsvRanges(qx, advisoryPackageId, [ + { + advisoryPackageId, + introducedVersion: '1.0.0', + fixedVersion: '1.2.0', + lastAffected: null, + }, + // '2.0.0'..'2.5.0' range dropped — OSV no longer reports it. + ]) + + const rows = await qx.select( + ` + SELECT introduced_version, deleted_at + FROM advisory_affected_ranges + WHERE advisory_package_id = $(advisoryPackageId) AND introduced_version = '2.0.0' + `, + { advisoryPackageId }, + ) + expect(rows).toHaveLength(1) + expect(rows[0].deleted_at).not.toBeNull() + }) + + it('revives a tombstoned tuple that reappears in a later sync', async () => { + await reconcileOsvRanges(qx, advisoryPackageId, [ + { + advisoryPackageId, + introducedVersion: '1.0.0', + fixedVersion: '1.2.0', + lastAffected: null, + }, + { + advisoryPackageId, + introducedVersion: '2.0.0', + fixedVersion: null, + lastAffected: '2.5.0', + }, + ]) + + const rows = await qx.select( + ` + SELECT deleted_at + FROM advisory_affected_ranges + WHERE advisory_package_id = $(advisoryPackageId) AND introduced_version = '2.0.0' + `, + { advisoryPackageId }, + ) + expect(rows).toHaveLength(1) + expect(rows[0].deleted_at).toBeNull() + }) + + it('supersedes a live deps.dev raw row once OSV owns the package', async () => { + await qx.result( + ` + INSERT INTO advisory_affected_ranges + (advisory_package_id, range_raw, unaffected_raw, introduced_version, created_at, updated_at) + VALUES ($(advisoryPackageId), '>=1.0.0 <1.2.0', NULL, NULL, NOW(), NOW()) + `, + { advisoryPackageId }, + ) + + await supersedeDepsDevRanges(qx, advisoryPackageId) + + const rows = await qx.select( + ` + SELECT deleted_at + FROM advisory_affected_ranges + WHERE advisory_package_id = $(advisoryPackageId) AND range_raw = '>=1.0.0 <1.2.0' + `, + { advisoryPackageId }, + ) + expect(rows).toHaveLength(1) + expect(rows[0].deleted_at).not.toBeNull() + }) +}) diff --git a/services/apps/packages_worker/src/osv/upsertAdvisory.ts b/services/apps/packages_worker/src/osv/upsertAdvisory.ts index bcede60f37..7e23b3c482 100644 --- a/services/apps/packages_worker/src/osv/upsertAdvisory.ts +++ b/services/apps/packages_worker/src/osv/upsertAdvisory.ts @@ -1,7 +1,9 @@ import { - deleteOsvOnlyRanges, + AdvisoryRangeInsertInput, findPackageId, - insertAdvisoryRange, + findRemovedAdvisoryPackageIds, + reconcileOsvRanges, + supersedeDepsDevRanges, upsertAdvisory, upsertAdvisoryPackage, } from '@crowd/data-access-layer/src/packages/osv' @@ -32,10 +34,14 @@ export function dedupeRanges(ranges: NormalizedRange[]): NormalizedRange[] { async function upsertOne(qx: QueryExecutor, record: NormalizedRecord): Promise { const { advisory, packages } = record - if (packages.length === 0) return const advisoryId = await upsertAdvisory(qx, advisory) + const entries: { + advisoryPackageId: number + ranges: AdvisoryRangeInsertInput[] + supersedeDepsDev: boolean + }[] = [] for (const entry of packages) { const packageId = await findPackageId(qx, entry.pkg) @@ -46,16 +52,57 @@ async function upsertOne(qx: QueryExecutor, record: NormalizedRecord): Promise ({ advisoryPackageId, introducedVersion: range.introducedVersion, fixedVersion: range.fixedVersion, lastAffected: range.lastAffected, - }) - } + })), + supersedeDepsDev: true, + }) + } + + // Packages OSV reported in a prior sync but that are missing from this payload + // (parseOsvRecord drops a package once it has no usable ranges left, and a + // corrected upstream record can drop one outright) never reach the loop above, + // so their previously live OSV ranges would otherwise sit forever as false + // positives and permanently block deps.dev's NOT EXISTS ownership guard. + // Reconcile them too, with an empty range set — reconcileOsvRanges soft-deletes + // every live OSV row when nothing in the new set matches. Not superseding + // deps.dev here: OSV no longer covers the package, so a live deps.dev row (if + // any) should stay live rather than be torn down alongside it. + const removedIds = await findRemovedAdvisoryPackageIds( + qx, + advisoryId, + entries.map((e) => e.advisoryPackageId), + ) + for (const advisoryPackageId of removedIds) { + entries.push({ advisoryPackageId, ranges: [], supersedeDepsDev: false }) + } + + // Lock advisory_packages rows in ascending id order — the same order deps.dev's + // bulk merge locks them in (ADVISORY_PACKAGES_LOCK_SQL's ORDER BY ap.id in + // ingestAdvisories.ts). OSV's payload order is otherwise arbitrary; locking out + // of order here would let this per-record transaction and a concurrent deps.dev + // chunk each hold one row and wait on the other's, deadlocking — and deps.dev's + // maximumAttempts: 1 turns a deadlock abort into a hard merge failure, not a retry. + entries.sort((a, b) => a.advisoryPackageId - b.advisoryPackageId) + + for (const { advisoryPackageId, ranges, supersedeDepsDev } of entries) { + // Row-locks this advisory_packages row (held until the enclosing transaction + // commits) before touching advisory_affected_ranges, matching the lock + // deps.dev's bulk merge takes on the same row — whichever writer locks first + // forces the other to wait and see its committed writes, closing the + // ownership race between the two independently-scheduled write paths + // (ADR-0001 §Write semantics). + await qx.result(`SELECT id FROM advisory_packages WHERE id = $(advisoryPackageId) FOR UPDATE`, { + advisoryPackageId, + }) + + await reconcileOsvRanges(qx, advisoryPackageId, ranges) + if (supersedeDepsDev) await supersedeDepsDevRanges(qx, advisoryPackageId) } } diff --git a/services/libs/data-access-layer/src/osspckgs/api.ts b/services/libs/data-access-layer/src/osspckgs/api.ts index 8b95019514..29e80ded57 100644 --- a/services/libs/data-access-layer/src/osspckgs/api.ts +++ b/services/libs/data-access-layer/src/osspckgs/api.ts @@ -1009,7 +1009,7 @@ export async function getAdvisoriesByPackageId( ${ADVISORY_RESOLUTION_EXPR} AS resolution FROM advisory_packages ap JOIN advisories a ON a.id = ap.advisory_id - LEFT JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id + LEFT JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id AND ar.deleted_at IS NULL JOIN packages p ON p.id = ap.package_id WHERE ap.package_id = $(packageId)::bigint GROUP BY a.osv_id, a.severity, a.is_critical, p.latest_version @@ -1117,7 +1117,7 @@ export async function getAdvisoriesByPurls( FROM packages p LEFT JOIN advisory_packages ap ON ap.package_id = p.id LEFT JOIN advisories a ON a.id = ap.advisory_id - LEFT JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id + LEFT JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id AND ar.deleted_at IS NULL WHERE p.purl = ANY($(purls)) GROUP BY p.purl, p.latest_version, a.osv_id, a.severity, a.is_critical ORDER BY diff --git a/services/libs/data-access-layer/src/packages/osv.ts b/services/libs/data-access-layer/src/packages/osv.ts index 8ce9c45b84..704bd1df10 100644 --- a/services/libs/data-access-layer/src/packages/osv.ts +++ b/services/libs/data-access-layer/src/packages/osv.ts @@ -139,42 +139,131 @@ export async function upsertAdvisoryPackage( return row.id as number } -// Delete every advisory_affected_ranges row for this advisory_package that -// lacks deps.dev raw columns — i.e. every OSV-owned row. The deps.dev BQ -// worker (future) is expected to populate range_raw / unaffected_raw on rows -// of its own and never set the structured introduced/fixed/last_affected -// columns; this predicate scopes the wipe to the OSV pipeline's rows so a -// deps.dev row is never clobbered on resync. OSV rows where all three -// structured columns are NULL (e.g. some MAL- "always vulnerable" ranges) -// are still deleted here and re-inserted from the new payload below — that's -// fine, the row is OSV-owned and idempotent. -export async function deleteOsvOnlyRanges( +// advisory_packages rows OSV previously wrote for this advisory that are absent +// from the current payload. parseOsvRecord drops a package once it has no +// usable ranges left, and a corrected upstream record can drop a package +// outright — either way the package never reaches upsertOne's per-entry loop, +// so its previously live OSV ranges would sit as permanent false positives and +// block deps.dev's NOT EXISTS ownership guard forever unless reconciled as a +// removal (see upsertOne in services/apps/packages_worker/src/osv/upsertAdvisory.ts). +export async function findRemovedAdvisoryPackageIds( + qx: QueryExecutor, + advisoryId: number, + presentAdvisoryPackageIds: number[], +): Promise { + const rows = await qx.select( + ` + SELECT id + FROM advisory_packages + WHERE advisory_id = $(advisoryId) + AND NOT (id = ANY($(presentAdvisoryPackageIds)::bigint[])) + `, + { advisoryId, presentAdvisoryPackageIds }, + ) + return rows.map((r: { id: number }) => r.id) +} + +// Diff-based upsert + soft-delete for OSV-owned advisory_affected_ranges rows. +// Replaces the old hard-delete + reinsert sweep (which minted a new PK for +// every unchanged range every sync, producing zombie rows once Sequin +// replicates DELETEs into a Tinybird ReplacingMergeTree that has no +// sign/is_deleted column — see ADR-0001 §`advisory_affected_ranges` +// delete/dedup strategy, CM-1258). +// +// An advisory_package whose range set hasn't changed since the last sync now +// produces zero writes: matching tuples are left untouched (not even an +// `updated_at` bump), so no Sequin/Tinybird event fires for them either. +// +// Tuple in both old and new, already a clean live OSV row -> untouched. +// Tuple only in new, or matching a live deps.dev raw row on the same key -> +// upserted (INSERT ON CONFLICT; clears deleted_at in case it was previously +// soft-deleted, and clears range_raw/unaffected_raw to reclaim the row from +// deps.dev — OSV wins on key overlap per ADR-0001 §Write semantics). Tuple +// only in old -> soft-deleted. +// Soft-delete (not hard-delete) is required because the unique key is the +// value tuple itself: when OSV corrects a range, the corrected tuple has no +// successor row to collapse into, so without deleted_at the stale tuple +// would sit forever as a false-positive vulnerable-range match. +export async function reconcileOsvRanges( qx: QueryExecutor, advisoryPackageId: number, + ranges: AdvisoryRangeInsertInput[], ): Promise { + const values = ranges.map((r) => ({ + introduced_version: r.introducedVersion, + fixed_version: r.fixedVersion, + last_affected: r.lastAffected, + })) + await qx.result( ` - DELETE FROM advisory_affected_ranges - WHERE advisory_package_id = $(advisoryPackageId) - AND range_raw IS NULL - AND unaffected_raw IS NULL + INSERT INTO advisory_affected_ranges + (advisory_package_id, introduced_version, fixed_version, last_affected, created_at, updated_at) + SELECT $(advisoryPackageId), v.introduced_version, v.fixed_version, v.last_affected, NOW(), NOW() + FROM jsonb_to_recordset($(values)::jsonb) AS v( + introduced_version text, + fixed_version text, + last_affected text + ) + ON CONFLICT (advisory_package_id, COALESCE(introduced_version, ''), COALESCE(fixed_version, ''), COALESCE(last_affected, '')) + DO UPDATE SET + updated_at = NOW(), + deleted_at = NULL, + range_raw = NULL, + unaffected_raw = NULL + WHERE advisory_affected_ranges.deleted_at IS NOT NULL + OR advisory_affected_ranges.range_raw IS NOT NULL + OR advisory_affected_ranges.unaffected_raw IS NOT NULL `, - { advisoryPackageId }, + { advisoryPackageId, values: JSON.stringify(values) }, + ) + + // Soft-delete OSV-owned live rows whose tuple isn't in the new set anymore. + await qx.result( + ` + UPDATE advisory_affected_ranges ar + SET deleted_at = NOW(), + updated_at = NOW() + WHERE ar.advisory_package_id = $(advisoryPackageId) + AND ar.deleted_at IS NULL + AND ar.range_raw IS NULL + AND ar.unaffected_raw IS NULL + AND NOT EXISTS ( + SELECT 1 + FROM jsonb_to_recordset($(values)::jsonb) AS v( + introduced_version text, + fixed_version text, + last_affected text + ) + WHERE COALESCE(v.introduced_version, '') = COALESCE(ar.introduced_version, '') + AND COALESCE(v.fixed_version, '') = COALESCE(ar.fixed_version, '') + AND COALESCE(v.last_affected, '') = COALESCE(ar.last_affected, '') + ) + `, + { advisoryPackageId, values: JSON.stringify(values) }, ) } -export async function insertAdvisoryRange( +// OSV is the source of truth for ranges over time (see ADR-0001 §Write +// semantics `advisories` row): once OSV has written its structured ranges +// for an advisory_package, any live deps.dev raw row for that same +// advisory_package is superseded and should stop being treated as current. +// Soft-deleted (not hard-deleted) for the same reason as reconcileOsvRanges — +// no successor row exists to collapse a deps.dev tuple into. +export async function supersedeDepsDevRanges( qx: QueryExecutor, - range: AdvisoryRangeInsertInput, + advisoryPackageId: number, ): Promise { await qx.result( ` - INSERT INTO advisory_affected_ranges - (advisory_package_id, introduced_version, fixed_version, last_affected, created_at, updated_at) - VALUES - ($(advisoryPackageId), $(introducedVersion), $(fixedVersion), $(lastAffected), NOW(), NOW()) + UPDATE advisory_affected_ranges + SET deleted_at = NOW(), + updated_at = NOW() + WHERE advisory_package_id = $(advisoryPackageId) + AND deleted_at IS NULL + AND (range_raw IS NOT NULL OR unaffected_raw IS NOT NULL) `, - range, + { advisoryPackageId }, ) } @@ -232,7 +321,7 @@ export async function getRangesForPackages(qx: QueryExecutor, ids: number[]): Pr a.osv_id FROM advisory_packages ap JOIN advisories a ON a.id = ap.advisory_id - JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id + JOIN advisory_affected_ranges ar ON ar.advisory_package_id = ap.id AND ar.deleted_at IS NULL WHERE ap.package_id IN ($(ids:csv)) AND (a.is_critical = TRUE OR a.osv_id LIKE 'MAL-%') `, diff --git a/services/libs/tinybird/datasources/advisoryAffectedRanges.datasource b/services/libs/tinybird/datasources/advisoryAffectedRanges.datasource index 3bfaa5ec84..8fd6282ab8 100644 --- a/services/libs/tinybird/datasources/advisoryAffectedRanges.datasource +++ b/services/libs/tinybird/datasources/advisoryAffectedRanges.datasource @@ -10,6 +10,7 @@ DESCRIPTION > - `rangeRaw` is the raw AffectedVersions string from the deps.dev BigQuery source (empty string if OSV-sourced). - `unaffectedRaw` is the raw UnaffectedVersions string from deps.dev BigQuery (empty string if OSV-sourced). - `createdAt` and `updatedAt` are row-level audit timestamps for Tinybird watermark-based sync. + - `deletedAt` is set when Postgres soft-deletes the row (stale OSV tuple or superseded deps.dev raw row); NULL means live. `FINAL` alone doesn't collapse this — a soft-deleted row and its live replacement are distinct PKs, so consumers must also filter `deletedAt IS NULL` (see ADR-0001 §`advisory_affected_ranges` delete/dedup strategy, CM-1258). SCHEMA > `id` UInt64 `json:$.record.id`, @@ -20,7 +21,8 @@ SCHEMA > `rangeRaw` String `json:$.record.range_raw` DEFAULT '', `unaffectedRaw` String `json:$.record.unaffected_raw` DEFAULT '', `createdAt` DateTime64(3) `json:$.record.created_at`, - `updatedAt` DateTime64(3) `json:$.record.updated_at` + `updatedAt` DateTime64(3) `json:$.record.updated_at`, + `deletedAt` Nullable(DateTime64(3)) `json:$.record.deleted_at` ENGINE ReplacingMergeTree ENGINE_PARTITION_KEY toYear(createdAt) diff --git a/services/libs/tinybird/pipes/ossPackages_enriched.pipe b/services/libs/tinybird/pipes/ossPackages_enriched.pipe index 6140ae63d0..2c1c4f830a 100644 --- a/services/libs/tinybird/pipes/ossPackages_enriched.pipe +++ b/services/libs/tinybird/pipes/ossPackages_enriched.pipe @@ -92,7 +92,8 @@ SQL > FROM ossPackages_enriched_packages AS p INNER JOIN advisoryPackages AS ap FINAL ON ap.packageId = p.id INNER JOIN advisories AS a FINAL ON a.id = ap.advisoryId - INNER JOIN advisoryAffectedRanges AS aar FINAL ON aar.advisoryPackageId = ap.id + INNER JOIN + advisoryAffectedRanges AS aar FINAL ON aar.advisoryPackageId = ap.id AND aar.deletedAt IS NULL LEFT JOIN (SELECT packageId, number, publishedAt AS introducedAt FROM versions FINAL) AS vi ON vi.packageId = p.id