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,15 @@ 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 DO NOTHING 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.
 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,9 +99,34 @@ 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
+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 NOTHING
 `
 
+// 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 = [
   'osv_id',
   'source',
@@ -259,6 +292,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/upsertAdvisory.ts

@@ -1,7 +1,7 @@
 import {
-  deleteOsvOnlyRanges,
   findPackageId,
-  insertAdvisoryRange,
+  reconcileOsvRanges,
+  supersedeDepsDevRanges,
   upsertAdvisory,
   upsertAdvisoryPackage,
 } from '@crowd/data-access-layer/src/packages/osv'
@@ -46,16 +46,27 @@ async function upsertOne(qx: QueryExecutor, record: NormalizedRecord): Promise<v
       packageName: entry.pkg.packageName,
     })
 
-    await deleteOsvOnlyRanges(qx, advisoryPackageId)
+    // 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 (ADVISORY_PACKAGES_LOCK_SQL in
+    // ingestAdvisories.ts) — 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,
+    })
 
-    for (const range of dedupeRanges(entry.ranges)) {
-      await insertAdvisoryRange(qx, {
+    await reconcileOsvRanges(
+      qx,
+      advisoryPackageId,
+      dedupeRanges(entry.ranges).map((range) => ({
         advisoryPackageId,
         introducedVersion: range.introducedVersion,
         fixedVersion: range.fixedVersion,
         lastAffected: range.lastAffected,
-      })
-    }
+      })),
+    )
+    await supersedeDepsDevRanges(qx, advisoryPackageId)
   }
 }
 

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