fix: reconcile OSV-dropped packages as range removals (CM-1258)

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

services/apps/packages_worker/src/osv/upsertAdvisory.ts

@@ -1,6 +1,7 @@
 import {
   AdvisoryRangeInsertInput,
   findPackageId,
+  findRemovedAdvisoryPackageIds,
   reconcileOsvRanges,
   supersedeDepsDevRanges,
   upsertAdvisory,
@@ -33,11 +34,14 @@ export function dedupeRanges(ranges: NormalizedRange[]): NormalizedRange[] {
 
 async function upsertOne(qx: QueryExecutor, record: NormalizedRecord): Promise<void> {
   const { advisory, packages } = record
-  if (packages.length === 0) return
 
   const advisoryId = await upsertAdvisory(qx, advisory)
 
-  const entries: { advisoryPackageId: number; ranges: AdvisoryRangeInsertInput[] }[] = []
+  const entries: {
+    advisoryPackageId: number
+    ranges: AdvisoryRangeInsertInput[]
+    supersedeDepsDev: boolean
+  }[] = []
   for (const entry of packages) {
     const packageId = await findPackageId(qx, entry.pkg)
 
@@ -56,9 +60,28 @@ async function upsertOne(qx: QueryExecutor, record: NormalizedRecord): Promise<v
         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
@@ -67,7 +90,7 @@ async function upsertOne(qx: QueryExecutor, record: NormalizedRecord): Promise<v
   // 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 } of entries) {
+  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
@@ -79,7 +102,7 @@ async function upsertOne(qx: QueryExecutor, record: NormalizedRecord): Promise<v
     })
 
     await reconcileOsvRanges(qx, advisoryPackageId, ranges)
-    await supersedeDepsDevRanges(qx, advisoryPackageId)
+    if (supersedeDepsDev) await supersedeDepsDevRanges(qx, advisoryPackageId)
   }
 }
 

services/libs/data-access-layer/src/packages/osv.ts

@@ -139,6 +139,30 @@ export async function upsertAdvisoryPackage(
   return row.id as number
 }
 
+// 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<number[]> {
+  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