fix: keep contacts from succeeded sources when an extractor fails [CM-1341] (#4362)

Signed-off-by: Mouad BANI <mouad-mb@outlook.com>

Author: mbani01

services/apps/packages_worker/src/security-contacts/ingestSingle.ts

@@ -94,8 +94,10 @@ export async function ingestSecurityContactsForPurl(
   const baseDeps = buildBaseDeps(config)
   try {
     const result = await processRepo(target, baseDeps, cdpQx)
-    if (result.status === 'ok') {
-      await writeContacts(qx, result.repoId, result.contacts, result.policies)
+    if (result.status === 'ok' || result.status === 'partial') {
+      await writeContacts(qx, result.repoId, result.contacts, result.policies, {
+        merge: result.status === 'partial',
+      })
     } else {
       await markRepoAttempted(qx, result.repoId)
     }

services/apps/packages_worker/src/security-contacts/processBatch.ts

@@ -135,13 +135,16 @@ export async function processRepo(
 
   const results = await Promise.allSettled(EXTRACTORS.map((extract) => extract(target, deps)))
 
-  // Write only when every extractor succeeded, so a failed one can't wipe contacts it didn't see.
-  const failed = results.find((r) => r.status === 'rejected') as PromiseRejectedResult | undefined
-  if (failed) {
+  let failedCount = 0
+  results.forEach((r, i) => {
+    if (r.status !== 'rejected') return
+    failedCount++
     log.warn(
-      { repoId: target.repoId, errMsg: failed.reason?.message },
-      'Extractor failed — preserving existing data',
+      { repoId: target.repoId, extractor: EXTRACTORS[i].name, errMsg: r.reason?.message },
+      'Extractor failed — keeping contacts from remaining sources',
     )
+  })
+  if (failedCount === results.length) {
     return { repoId: target.repoId, status: 'extractor-failed' }
   }
 
@@ -177,7 +180,12 @@ export async function processRepo(
   }
 
   const scored = reconcile(contacts)
-  return { repoId: target.repoId, status: 'ok', contacts: scored, policies }
+  return {
+    repoId: target.repoId,
+    status: failedCount > 0 ? 'partial' : 'ok',
+    contacts: scored,
+    policies,
+  }
 }
 
 export async function processBatch(
@@ -228,10 +236,12 @@ export async function processBatch(
 
     const extractionDurationMs = Date.now() - extractionStartedAt
     const ok = outcomes.filter((o) => o.status === 'ok').length
+    const partial = outcomes.filter((o) => o.status === 'partial').length
     log.info(
       {
         ok,
-        failed: outcomes.length - ok,
+        partial,
+        failed: outcomes.length - ok - partial,
         durationMs: extractionDurationMs,
         reposPerSec: Number((outcomes.length / (extractionDurationMs / 1000)).toFixed(1)),
       },

services/apps/packages_worker/src/security-contacts/types.ts

@@ -81,4 +81,10 @@ export type Extractor = (target: RepoTarget, deps: ExtractorDeps) => Promise<Ext
 
 export type ProcessRepoResult =
   | { repoId: string; status: 'ok'; contacts: ScoredContact[]; policies: Partial<RepoPolicies> }
+  | {
+      repoId: string
+      status: 'partial'
+      contacts: ScoredContact[]
+      policies: Partial<RepoPolicies>
+    }
   | { repoId: string; status: 'extractor-failed' }

services/apps/packages_worker/src/security-contacts/writeContacts.ts

@@ -57,17 +57,23 @@ export async function markRepoAttempted(qx: QueryExecutor, repoId: string): Prom
 //
 // Soft-delete: mark every active row stale, then upsert this pass's contacts, reviving whatever
 // was rediscovered. Readers of this table must filter on deleted_at IS NULL.
+//
+// merge: skip the soft-delete when this pass ran with a failed extractor — a failed source
+// can't wipe contacts it didn't see; stale rows are cleaned on the next fully-successful pass.
 export async function writeContacts(
   qx: QueryExecutor,
   repoId: string,
   contacts: ScoredContact[],
   policies: Partial<RepoPolicies>,
+  opts: { merge?: boolean } = {},
 ): Promise<void> {
   await qx.tx(async (tx) => {
-    await tx.result(
-      'UPDATE security_contacts SET deleted_at = NOW(), updated_at = NOW() WHERE repo_id = $(repoId) AND deleted_at IS NULL',
-      { repoId },
-    )
+    if (!opts.merge) {
+      await tx.result(
+        'UPDATE security_contacts SET deleted_at = NOW(), updated_at = NOW() WHERE repo_id = $(repoId) AND deleted_at IS NULL',
+        { repoId },
+      )
+    }
 
     if (contacts.length > 0) {
       await tx.result(
@@ -114,12 +120,12 @@ export async function markReposAttempted(qx: QueryExecutor, repoIds: string[]):
   )
 }
 
-type OkResult = Extract<ProcessRepoResult, { status: 'ok' }>
+type PersistableResult = Extract<ProcessRepoResult, { status: 'ok' | 'partial' }>
 
 // Bounds blast radius: a bad row only forces a re-extract of its chunk next sweep, not the whole batch.
 const WRITE_CHUNK_SIZE = 100
 
-function prepareBulkPolicyUpdate(chunk: OkResult[]): string {
+function prepareBulkPolicyUpdate(chunk: PersistableResult[]): string {
   const rows = chunk.map(
     (_, i) =>
       `($(id${i})::bigint, $(securityPolicyUrl${i}), $(vulnerabilityReportingUrl${i}), $(pvrResolved${i})::boolean, $(bugBountyUrl${i}), $(securityTxtUrl${i}), $(pvrEnabled${i})::boolean)`,
@@ -156,15 +162,20 @@ function prepareBulkPolicyUpdate(chunk: OkResult[]): string {
   )
 }
 
-async function writeContactsChunk(qx: QueryExecutor, chunk: OkResult[]): Promise<void> {
+async function writeContactsChunk(qx: QueryExecutor, chunk: PersistableResult[]): Promise<void> {
   if (chunk.length === 0) return
-  const repoIds = chunk.map((r) => r.repoId)
+
+  // Partial repos get a merge-only write: their failed extractor couldn't re-see existing
+  // contacts, so soft-deleting would wipe data the pass never evaluated.
+  const fullRefreshIds = chunk.filter((r) => r.status === 'ok').map((r) => r.repoId)
 
   await qx.tx(async (tx) => {
-    await tx.result(
-      'UPDATE security_contacts SET deleted_at = NOW(), updated_at = NOW() WHERE repo_id = ANY($(repoIds)::bigint[]) AND deleted_at IS NULL',
-      { repoIds },
-    )
+    if (fullRefreshIds.length > 0) {
+      await tx.result(
+        'UPDATE security_contacts SET deleted_at = NOW(), updated_at = NOW() WHERE repo_id = ANY($(repoIds)::bigint[]) AND deleted_at IS NULL',
+        { repoIds: fullRefreshIds },
+      )
+    }
 
     const rows = chunk.flatMap((r) => r.contacts.map((c) => toContactRow(r.repoId, c)))
     if (rows.length > 0) {
@@ -184,7 +195,9 @@ export async function writeContactsBatch(
   const attemptedOnlyIds = outcomes
     .filter((o) => o.status === 'extractor-failed')
     .map((o) => o.repoId)
-  const ok = outcomes.filter((o): o is OkResult => o.status === 'ok')
+  const ok = outcomes.filter(
+    (o): o is PersistableResult => o.status === 'ok' || o.status === 'partial',
+  )
 
   let firstError: Error | undefined
   for (let i = 0; i < ok.length; i += WRITE_CHUNK_SIZE) {