fix: stamp meta:fill on deps export reuse (CM-1296)

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

Author: themarolt

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

@@ -6,6 +6,7 @@ import {
   findExportedJobByGcsPrefix,
   findLatestExportedJobByKind,
   markJobStatus,
+  mergeJobTableRowCounts,
 } from '@crowd/data-access-layer'
 import { getServiceChildLogger } from '@crowd/logging'
 
@@ -86,6 +87,10 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport
           { jobKind, exportName, jobId: prior.id, gcsPrefix: prior.gcsPrefix },
           'exportName match — skipping BQ, loading from named export',
         )
+        // A --fill-constraints run reusing an existing export must still stamp meta:fill on the
+        // reused row, else a later --resume-job reads fill=false and reverts the merge to
+        // ON CONFLICT DO NOTHING — silently skipping the version_constraint backfill.
+        if (isFill) await mergeJobTableRowCounts(qx, prior.id, { 'meta:fill': 1 })
         return {
           gcsPrefix: prior.gcsPrefix,
           rowCount: prior.rowCountBq,
@@ -143,6 +148,9 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport
           { jobKind, jobId: prior.id, gcsPrefix: prior.gcsPrefix },
           'reuseExports=true — skipping BQ, loading from prior export',
         )
+        // See named-export path above: persist meta:fill so a later --resume-job keeps the
+        // fill-constraints merge instead of reverting to ON CONFLICT DO NOTHING.
+        if (isFill) await mergeJobTableRowCounts(qx, prior.id, { 'meta:fill': 1 })
         return {
           gcsPrefix: prior.gcsPrefix,
           rowCount: prior.rowCountBq,
@@ -171,6 +179,9 @@ export async function bqExportToGcs(input: BqExportToGcsInput): Promise<BqExport
         { jobKind, jobId: existing.id, gcsPrefix },
         'GCS files already exist — reusing export',
       )
+      // Same-runId reuse (Temporal retry): the export row already has meta:fill from its own
+      // 'exporting' write, but re-stamp defensively so the fill flag can never be lost on reuse.
+      if (isFill) await mergeJobTableRowCounts(qx, existing.id, { 'meta:fill': 1 })
       return { gcsPrefix, rowCount: existing.rowCountBq, bqBytesBilled: 0, jobId: existing.id }
     }
   }