fix: dedup advisory_affected_ranges via diff-based upsert (CM-1258)

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;

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

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,

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<void> {
+  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<RangeRow[]> {
+  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()
+  })
+})