fix: resume deps ingest with original export settings (CM-1296)

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

Author: themarolt

services/apps/packages_worker/src/deps-dev/activities/bqExportToGcs.ts

@@ -32,6 +32,12 @@ export interface BqExportToGcsInput {
   // is replaced by a server-side maximumBytesBilled cap, since a dry-run only validates the first
   // statement and cannot predict the WHILE loop's total scan. See ADR-0004.
   isScript?: boolean
+  // Fill-constraints run (package_dependencies only): the export is a full Option-A scan but the
+  // downstream merge upserts version_constraint (ON CONFLICT DO UPDATE) instead of DO NOTHING. Full
+  // and fill produce identical parquet, so sync_mode alone can't tell them apart — persist it in the
+  // job meta so a --resume-job run reprocesses the export with the correct merge instead of silently
+  // reverting to DO NOTHING (which skips the backfill).
+  isFill?: boolean
 }
 
 export interface BqExportToGcsOutput {
@@ -53,6 +59,7 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport
     exportName,
     ecosystems,
     isScript,
+    isFill,
   } = input
 
   // Named exports use a stable GCS path independent of runId so they survive across bootstrap runs.
@@ -115,6 +122,7 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport
           tableRowCounts: {
             'bq:export': 0,
             ...(ecosystems ? { 'meta:ecosystems': ecosystems } : {}),
+            ...(isFill ? { 'meta:fill': 1 } : {}),
           },
         })
         return { gcsPrefix: namedPrefix, rowCount: 0, bqBytesBilled: 0, jobId }
@@ -212,9 +220,13 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport
   const provisionalDate = snapshotAt ? new Date(snapshotAt) : null
   const jobId = await createIngestJob(qx, jobKind, syncMode, provisionalDate, exportName)
 
-  // H7: mark exporting before we start the BQ job; store ecosystems filter in table_row_counts JSONB.
+  // H7: mark exporting before we start the BQ job; store ecosystems filter + fill flag in the
+  // table_row_counts JSONB so --resume-job can restore the original export's settings.
+  const exportMeta: Record<string, string | number | string[]> = {}
+  if (ecosystems) exportMeta['meta:ecosystems'] = ecosystems
+  if (isFill) exportMeta['meta:fill'] = 1
   await markJobStatus(qx, jobId, 'exporting', {
-    ...(ecosystems ? { tableRowCounts: { 'meta:ecosystems': ecosystems } } : {}),
+    ...(Object.keys(exportMeta).length > 0 ? { tableRowCounts: exportMeta } : {}),
   })
 
   // From here the row is 'exporting'; any BQ failure (incl. script-mode maximumBytesBilled aborts)

services/apps/packages_worker/src/deps-dev/activities/getResumeExport.ts

@@ -15,6 +15,8 @@ export interface GetResumeExportOutput {
   progressDone: number
   progressTotal: number
   rowCountPg: number
+  ecosystems: string[] | null
+  fill: boolean
 }
 
 // Pure fetch of a prior job's resume-relevant fields (export path, status, file-load progress,
@@ -38,5 +40,7 @@ export async function getResumeExport(
     progressDone: job.progressDone,
     progressTotal: job.progressTotal,
     rowCountPg: job.rowCountPg,
+    ecosystems: job.ecosystems,
+    fill: job.fill,
   }
 }

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

@@ -183,8 +183,9 @@ export async function ingestDependencies(opts: {
   // and restart the chunk loop where its staging load left off. Idempotent modes only (see gate).
   resumeJobId?: number
 }): Promise<{ rowCountBq: number }> {
-  const ecosystems = opts.ecosystems ?? DEPS_DEFAULT_ECOSYSTEMS
-  const isFill = opts.fillConstraints === true
+  // let: on resume these are overridden with the prior export's stored settings (see resume block).
+  let ecosystems = opts.ecosystems ?? DEPS_DEFAULT_ECOSYSTEMS
+  let isFill = opts.fillConstraints === true
   // Fill mode forces Option A — Option B selects NULL for version_constraint, making the fill a no-op.
   const tableOption = isFill ? 'A' : (opts.depsTableOption ?? 'A')
   // Full/fill scan the *Latest views; incremental reads the `today` partition. Drives both the SQL
@@ -234,6 +235,13 @@ export async function ingestDependencies(opts: {
     exportResult = { jobId: prior.jobId, gcsPrefix: prior.gcsPrefix, rowCount: 0 }
     resumeProgressDone = prior.progressDone
     resumeRowCountPg = prior.rowCountPg
+    // Reprocess the export with its ORIGINAL settings, not the current CLI opts. A narrower
+    // ecosystems list would drop versions-lookup rows (deps silently unmerged); the wrong fill flag
+    // would pick ON CONFLICT DO NOTHING and skip the version_constraint backfill. Pre-meta jobs (no
+    // stored ecosystems) fall back to all ecosystems — a superset lookup never drops rows. fill is
+    // OR'd with the current flag so an explicit --fill-constraints still works for pre-meta jobs.
+    ecosystems = prior.ecosystems ?? DEPS_DEFAULT_ECOSYSTEMS
+    isFill = prior.fill || isFill
   } else {
     // Guard against corrupt deps.dev resolved-graph snapshots BEFORE the (multi-hour) export.
     // Skip when reusing a prior export — we're re-importing already-validated parquet, not
@@ -284,6 +292,8 @@ export async function ingestDependencies(opts: {
       reuseExports: opts.reuseExports,
       exportName: opts.exportName,
       ecosystems,
+      // Persist fill in the job meta so a later --resume-job restores the upsert merge (see resume block).
+      isFill,
     })
   }
 

services/libs/data-access-layer/src/osspckgs/ingestJobs.ts

@@ -104,12 +104,20 @@ export async function getIngestJobForResume(
   progressDone: number
   progressTotal: number
   rowCountPg: number
+  // Original export parameters, so resume reprocesses the export with the SAME settings rather than
+  // whatever the current CLI opts happen to be: ecosystems (drives the versions-lookup filter) and
+  // fill (drives the merge SQL — fill upserts version_constraint, non-fill is ON CONFLICT DO NOTHING).
+  // ecosystems is null for jobs exported before the meta key existed; fill defaults to false.
+  ecosystems: string[] | null
+  fill: boolean
 } | null> {
   const row = await qx.selectOneOrNone(
     `
     SELECT id, job_kind, status, sync_mode, gcs_prefix, row_count_pg,
            COALESCE((table_row_counts->>'progress:done')::int, 0)  AS progress_done,
-           COALESCE((table_row_counts->>'progress:total')::int, 0) AS progress_total
+           COALESCE((table_row_counts->>'progress:total')::int, 0) AS progress_total,
+           table_row_counts->'meta:ecosystems'                     AS ecosystems,
+           COALESCE((table_row_counts->>'meta:fill')::bool, false)  AS fill
     FROM osspckgs_ingest_jobs
     WHERE id = $(id)
     `,
@@ -125,6 +133,8 @@ export async function getIngestJobForResume(
         progressDone: Number(row.progress_done ?? 0),
         progressTotal: Number(row.progress_total ?? 0),
         rowCountPg: Number(row.row_count_pg ?? 0),
+        ecosystems: (row.ecosystems as string[] | null) ?? null,
+        fill: Boolean(row.fill),
       }
     : null
 }