fix: revive soft-deleted ranges and fix lock ordering deadlock (CM-1258)

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

Author: themarolt

services/apps/packages_worker/src/deps-dev/workflows/ingestAdvisories.ts

@@ -82,10 +82,14 @@ ON CONFLICT (advisory_id, ecosystem, package_name) DO NOTHING
 // 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.
+// 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
@@ -106,7 +110,13 @@ WHERE NOT EXISTS (
     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
+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

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

@@ -1,4 +1,5 @@
 import {
+  AdvisoryRangeInsertInput,
   findPackageId,
   reconcileOsvRanges,
   supersedeDepsDevRanges,
@@ -36,6 +37,7 @@ async function upsertOne(qx: QueryExecutor, record: NormalizedRecord): Promise<v
 
   const advisoryId = await upsertAdvisory(qx, advisory)
 
+  const entries: { advisoryPackageId: number; ranges: AdvisoryRangeInsertInput[] }[] = []
   for (const entry of packages) {
     const packageId = await findPackageId(qx, entry.pkg)
 
@@ -46,26 +48,37 @@ async function upsertOne(qx: QueryExecutor, record: NormalizedRecord): Promise<v
       packageName: entry.pkg.packageName,
     })
 
-    // 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,
-    })
-
-    await reconcileOsvRanges(
-      qx,
+    entries.push({
       advisoryPackageId,
-      dedupeRanges(entry.ranges).map((range) => ({
+      ranges: dedupeRanges(entry.ranges).map((range) => ({
         advisoryPackageId,
         introducedVersion: range.introducedVersion,
         fixedVersion: range.fixedVersion,
         lastAffected: range.lastAffected,
       })),
-    )
+    })
+  }
+
+  // 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 } 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)
     await supersedeDepsDevRanges(qx, advisoryPackageId)
   }
 }