fix: isolate per-repo failures in security contacts batch

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

Author: mbani01

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

@@ -107,23 +107,25 @@ async function processRepo(
   deps: ExtractorDeps,
   qx: QueryExecutor,
 ): Promise<void> {
-  // One failing extractor must not sink the repo.
   const results = await Promise.allSettled(EXTRACTORS.map((extract) => extract(target, deps)))
 
-  // Every extractor failed (transient outage) — preserve existing data, just record the attempt.
-  if (results.every((r) => r.status === 'rejected')) {
-    log.warn({ repoId: target.repoId }, 'All extractors failed — preserving existing data')
+  // Replace the repo's contacts only when every extractor succeeded. If any failed (transient
+  // error — non-200s throw), a destructive rewrite would drop contacts a failed tier-A/B extractor
+  // still has, so preserve existing data and just record the attempt; retried next cadence.
+  const failed = results.find((r) => r.status === 'rejected') as PromiseRejectedResult | undefined
+  if (failed) {
+    log.warn(
+      { repoId: target.repoId, errMsg: failed.reason?.message },
+      'Extractor failed — preserving existing data',
+    )
     await markRepoAttempted(qx, target.repoId)
     return
   }
 
   let contacts: RawContact[] = []
   const policies: Partial<RepoPolicies> = {}
   for (const r of results) {
-    if (r.status !== 'fulfilled') {
-      log.warn({ repoId: target.repoId, errMsg: r.reason?.message }, 'Extractor failed')
-      continue
-    }
+    if (r.status !== 'fulfilled') continue
     contacts.push(...r.value.contacts)
     for (const [key, value] of Object.entries(r.value.policies)) {
       if (!(policies as Record<string, unknown>)[key] && value != null) {
@@ -153,7 +155,16 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise<B
   }
 
   const targets = batch.map(toTarget)
-  await mapWithConcurrency(targets, CONCURRENCY, (target) => processRepo(target, deps, qx))
+  // Isolate per-repo failures: mapWithConcurrency is fail-fast, so a single repo's DB error must
+  // not abort the batch (and, across Temporal retries, halt the whole sweep). Unmarked repos on
+  // error are simply retried next sweep.
+  await mapWithConcurrency(targets, CONCURRENCY, async (target) => {
+    try {
+      await processRepo(target, deps, qx)
+    } catch (err) {
+      log.error({ repoId: target.repoId, errMsg: (err as Error).message }, 'Repo processing failed')
+    }
+  })
 
   log.info({ processed: targets.length }, 'Security contacts batch complete')
   return { processed: targets.length }

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

@@ -2,11 +2,6 @@ import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
 
 import { RepoPolicies, ScoredContact } from './types'
 
-/**
- * Idempotent per-repo recompute: replace the repo's security_contacts rows and refresh
- * the policy columns in one transaction. Policy columns use COALESCE so a run that doesn't
- * rediscover a field (partial/failed extractor pass) never clears a previously known value.
- */
 // Records the attempt without touching contacts/policies — preserves existing data on total
 // failure while advancing contacts_last_refreshed so the repo isn't reprocessed this sweep.
 export async function markRepoAttempted(qx: QueryExecutor, repoId: string): Promise<void> {
@@ -15,6 +10,11 @@ export async function markRepoAttempted(qx: QueryExecutor, repoId: string): Prom
   })
 }
 
+/**
+ * Idempotent per-repo recompute: replace the repo's security_contacts rows and refresh
+ * the policy columns in one transaction. Policy columns use COALESCE so a run that doesn't
+ * rediscover a field (partial/failed extractor pass) never clears a previously known value.
+ */
 export async function writeContacts(
   qx: QueryExecutor,
   repoId: string,