fix: wrap packagist metadata writes in a transaction

Signed-off-by: anilb <epipav@gmail.com>

Author: epipav

services/apps/packages_worker/src/packagist/__tests__/persistMetadata.test.ts

@@ -23,7 +23,9 @@ const mockVersions = vi.mocked(upsertPackagistVersions)
 const mockIds = vi.mocked(getPackagistPackageIdsByNames)
 const mockDeps = vi.mocked(reconcileVersionDependencies)
 
-const qx = {} as QueryExecutor
+const qx = {
+  tx: vi.fn((cb: (t: QueryExecutor) => Promise<void>) => cb(qx)),
+} as unknown as QueryExecutor
 const PURL = 'pkg:composer/monolog/monolog'
 
 const expanded: PackagistExpandedVersion[] = [

services/apps/packages_worker/src/packagist/upsertMetadata.ts

@@ -28,77 +28,85 @@ export async function persistPackagistMetadata(
   const { versionRows, latestVersion, firstReleaseAt, latestReleaseAt, licenses, homepage } =
     buildPackagistVersionRows(expanded)
 
-  const agg = await updatePackagistVersionAggregates(qx, purl, {
-    versionsCount: versionRows.length,
-    latestVersion,
-    firstReleaseAt,
-    latestReleaseAt,
-    licenses,
-    homepage,
-  })
+  let found = false
+  const changedFields: string[] = []
+  let unresolvedDependencyTargets = 0
 
-  if (!agg) {
-    return { found: false, changedFields: [], unresolvedDependencyTargets: 0 }
-  }
+  // One transaction for the whole p2 write path: the aggregates update, version
+  // upsert + is_latest cleanup, and dependency reconcile (delete stale + upsert
+  // current) are each multi-statement on their own — sharing one tx means a failure
+  // partway through can never leave versions/dependencies half-refreshed.
+  await qx.tx(async (t) => {
+    const agg = await updatePackagistVersionAggregates(t, purl, {
+      versionsCount: versionRows.length,
+      latestVersion,
+      firstReleaseAt,
+      latestReleaseAt,
+      licenses,
+      homepage,
+    })
 
-  const changedFields = [...agg.changedFields]
+    if (!agg) return
 
-  let versionIds: Array<{ number: string; id: string }> = []
-  if (versionRows.length > 0) {
-    const versionResult = await upsertPackagistVersions(qx, agg.id, versionRows, latestVersion)
-    changedFields.push(...versionResult.changedFields)
-    versionIds = versionResult.versionIds
-  }
+    found = true
+    changedFields.push(...agg.changedFields)
 
-  // One pass over the tagged versions collects target names and provisional edges;
-  // target ids are then resolved in a single batch query.
-  const versionMap = new Map(versionIds.map((v) => [v.number, v.id]))
-  const targetNames = new Set<string>()
-  const pending: Array<{ versionId: string; dep: PackagistDependency }> = []
+    let versionIds: Array<{ number: string; id: string }> = []
+    if (versionRows.length > 0) {
+      const versionResult = await upsertPackagistVersions(t, agg.id, versionRows, latestVersion)
+      changedFields.push(...versionResult.changedFields)
+      versionIds = versionResult.versionIds
+    }
 
-  for (const v of expanded) {
-    if (isPackagistDevVersion(v.version, v.version_normalized)) continue
-    const versionId = versionMap.get(v.version)
-    if (!versionId) continue
+    // One pass over the tagged versions collects target names and provisional edges;
+    // target ids are then resolved in a single batch query.
+    const versionMap = new Map(versionIds.map((v) => [v.number, v.id]))
+    const targetNames = new Set<string>()
+    const pending: Array<{ versionId: string; dep: PackagistDependency }> = []
 
-    for (const dep of extractVersionDependencies(v)) {
-      targetNames.add(dep.name)
-      pending.push({ versionId, dep })
+    for (const v of expanded) {
+      if (isPackagistDevVersion(v.version, v.version_normalized)) continue
+      const versionId = versionMap.get(v.version)
+      if (!versionId) continue
+
+      for (const dep of extractVersionDependencies(v)) {
+        targetNames.add(dep.name)
+        pending.push({ versionId, dep })
+      }
     }
-  }
 
-  // Reconciled for every version being refreshed (not just ones with a resolved
-  // dependency this pass) — a version whose manifest drops a requirement, or all of
-  // them, must have its stale package_dependencies rows removed, not just the newly
-  // declared ones upserted.
-  let unresolvedDependencyTargets = 0
-  if (versionIds.length > 0) {
-    const edges: VersionDependencyEdge[] = []
-    if (pending.length > 0) {
-      const idMap = await getPackagistPackageIdsByNames(qx, Array.from(targetNames))
-      for (const { versionId, dep } of pending) {
-        const dependsOnId = idMap.get(dep.name)
-        if (!dependsOnId) {
-          unresolvedDependencyTargets++
-          continue
+    // Reconciled for every version being refreshed (not just ones with a resolved
+    // dependency this pass) — a version whose manifest drops a requirement, or all of
+    // them, must have its stale package_dependencies rows removed, not just the newly
+    // declared ones upserted.
+    if (versionIds.length > 0) {
+      const edges: VersionDependencyEdge[] = []
+      if (pending.length > 0) {
+        const idMap = await getPackagistPackageIdsByNames(t, Array.from(targetNames))
+        for (const { versionId, dep } of pending) {
+          const dependsOnId = idMap.get(dep.name)
+          if (!dependsOnId) {
+            unresolvedDependencyTargets++
+            continue
+          }
+          edges.push({
+            packageId: agg.id,
+            versionId,
+            dependsOnId,
+            constraint: dep.constraint,
+            kind: dep.kind,
+          })
         }
-        edges.push({
-          packageId: agg.id,
-          versionId,
-          dependsOnId,
-          constraint: dep.constraint,
-          kind: dep.kind,
-        })
       }
-    }
 
-    const depChanges = await reconcileVersionDependencies(
-      qx,
-      versionIds.map((v) => v.id),
-      edges,
-    )
-    changedFields.push(...depChanges)
-  }
+      const depChanges = await reconcileVersionDependencies(
+        t,
+        versionIds.map((v) => v.id),
+        edges,
+      )
+      changedFields.push(...depChanges)
+    }
+  })
 
-  return { found: true, changedFields, unresolvedDependencyTargets }
+  return { found, changedFields, unresolvedDependencyTargets }
 }