fix: serialize OSV/deps.dev advisory range writes with row locks (CM-1258)

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

Author: themarolt

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

@@ -109,6 +109,24 @@ WHERE NOT EXISTS (
 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',
@@ -274,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

@@ -46,6 +46,16 @@ 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,
       advisoryPackageId,

services/libs/tinybird/pipes/ossPackages_enriched.pipe

@@ -92,8 +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
-        AND aar.deletedAt IS NULL
+    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