fix: restore merge dedup and relax deps resume guards (CM-1296)

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

Author: themarolt

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

@@ -93,6 +93,18 @@ export async function bootstrapOsspckgs(opts: {
   const activeKinds = opts.kinds ? new Set(opts.kinds) : null
   const runs = (kind: string) => !activeKinds || activeKinds.has(kind)
 
+  // Resume reuses a prior package_dependencies export and skips watermark/partition validation.
+  // Hard-enforce it targets ONLY package_dependencies so a stray resumeJobId can't silently run other
+  // kinds without their safety checks. The CLI validates this too; this is the fail-fast backstop.
+  if (
+    resume &&
+    !(activeKinds && activeKinds.size === 1 && activeKinds.has('package_dependencies'))
+  ) {
+    throw new ApplicationFailure(
+      'resumeJobId is only valid with kinds=[package_dependencies] — refusing to skip validation for other kinds',
+    )
+  }
+
   const jobKinds: JobKind[] = (
     ['packages', 'versions', 'package_dependencies', 'advisories', 'advisory_packages'] as JobKind[]
   ).filter((k) => runs(k))

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

@@ -76,7 +76,7 @@ INSERT INTO package_dependencies (
   package_id, version_id, depends_on_id, depends_on_version_id,
   version_constraint, dependency_kind, is_optional, created_at, updated_at
 )
-SELECT
+SELECT DISTINCT ON (pv.id, pd.id)
   pv.package_id, pv.id, pd.id, dv.id,
   sp.version_constraint, 'direct', FALSE, NOW(), NOW()
 FROM staging.osspckgs_deps_raw sp
@@ -100,6 +100,11 @@ JOIN packages pd ON pd.ecosystem = sp.ecosystem
       WHEN sp.to_name LIKE '@%/%'  THEN SPLIT_PART(sp.to_name, '/', 2)
       ELSE sp.to_name END
 LEFT JOIN versions dv ON dv.package_id = pd.id AND dv.number = sp.to_version
+-- DISTINCT ON collapses duplicate (root, dep) pairs BQ emits with different resolved to_version;
+-- ORDER BY keeps the highest to_version (best-resolvable depends_on_version_id), matching the fill
+-- variant. ON CONFLICT then handles cross-chunk / pre-existing rows. Full lost this when MERGE_SQL_FULL
+-- (which had the same DISTINCT ON) was removed with the drop/rebuild-index step.
+ORDER BY pv.id, pd.id, sp.to_version DESC NULLS LAST
 ON CONFLICT (version_id, depends_on_id, dependency_kind) DO NOTHING
 `
 

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

@@ -25,6 +25,11 @@ const { mergeStagingToTable } = proxyActivities<typeof depsDevActivities>({
   retry: { maximumAttempts: 1 },
 })
 
+const { setJobStep } = proxyActivities<typeof depsDevActivities>({
+  startToCloseTimeout: '30 seconds',
+  retry: { maximumAttempts: 3 },
+})
+
 const STAGING_TABLE = 'staging.osspckgs_versions_raw'
 
 const STAGING_DDL = `
@@ -46,11 +51,15 @@ INSERT INTO versions (
   package_id, ecosystem, namespace, name, number, published_at, is_prerelease, licenses, last_synced_at,
   created_at
 )
-SELECT
+SELECT DISTINCT ON (p.id, s.number)
   p.id, s.ecosystem, p.namespace, p.name, s.number, s.published_at, s.is_prerelease, s.licenses, NOW(),
   NOW()
 FROM staging.osspckgs_versions_raw s
 JOIN packages p ON p.purl = s.purl
+-- DISTINCT ON + ORDER BY keeps the most recently published row when BQ emits duplicate
+-- (purl, number) rows with differing published_at/licenses, so ON CONFLICT DO NOTHING no longer
+-- retains an arbitrary insert-order row.
+ORDER BY p.id, s.number, s.published_at DESC NULLS LAST
 ON CONFLICT (package_id, number) DO NOTHING
 `
 
@@ -127,6 +136,10 @@ export async function ingestVersions(opts: {
   let priorStagingRows = 0
   const priorTableRowCounts: Record<string, number> = {}
 
+  // Mark the phase before the loop so the monitor shows 'merging' instead of the stale prior step
+  // (e.g. 'loading') during the merge — mirrors ingestDependencies.
+  await setJobStep({ jobId: exportResult.jobId, step: 'merging' })
+
   for (let chunkIndex = 0; chunkIndex < totalChunks; chunkIndex++) {
     const start = chunkIndex * filesPerChunk
     const chunk = fileNames.slice(start, start + filesPerChunk)

services/apps/packages_worker/src/scripts/triggerBootstrap.ts

@@ -48,7 +48,8 @@ Options:
   --resume-job <id>      Resume a partially-merged package_dependencies job by id. Skips BQ,
                          reuses that job's exact parquet export, and restarts the chunk loop
                          where its staging load left off (idempotent ON CONFLICT overlap).
-                         Requires: --kinds package_dependencies AND (incremental | --fill-constraints).
+                         Requires: --kinds package_dependencies. Works in any mode (all deps
+                         merges are idempotent).
   --help                 Show this help
 
 Examples:
@@ -170,19 +171,14 @@ async function main(): Promise<void> {
     }
   }
 
-  // Resume is deps-only and merge-idempotent-only: it re-runs already-committed chunks, safe under
-  // incremental (ON CONFLICT DO NOTHING) or --fill-constraints (ON CONFLICT DO UPDATE), not a full load.
+  // Resume is deps-only: it re-runs already-committed chunks. Safe in every mode because all deps
+  // merges are now idempotent — full and incremental use ON CONFLICT DO NOTHING, --fill-constraints
+  // uses ON CONFLICT DO UPDATE. (Full became idempotent when the drop/rebuild-index step was removed.)
   if (resumeJobId !== undefined) {
     if (!kinds || kinds.length !== 1 || kinds[0] !== 'package_dependencies') {
       console.error('--resume-job requires: --kinds package_dependencies (resume is deps-only)')
       process.exit(1)
     }
-    if (mode !== 'incremental' && !fillConstraints) {
-      console.error(
-        '--resume-job requires incremental mode or --fill-constraints (idempotent merges only)',
-      )
-      process.exit(1)
-    }
   }
 
   const cfg = TEMPORAL_CONFIG()