diff --git a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts index 77cd5793fb..e75a722f57 100644 --- a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts +++ b/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) } diff --git a/services/apps/packages_worker/src/security-contacts/processBatch.ts b/services/apps/packages_worker/src/security-contacts/processBatch.ts index 2a66948a38..1c36b82c4f 100644 --- a/services/apps/packages_worker/src/security-contacts/processBatch.ts +++ b/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)), }, diff --git a/services/apps/packages_worker/src/security-contacts/types.ts b/services/apps/packages_worker/src/security-contacts/types.ts index 45cf60b1ab..734b804c64 100644 --- a/services/apps/packages_worker/src/security-contacts/types.ts +++ b/services/apps/packages_worker/src/security-contacts/types.ts @@ -81,4 +81,10 @@ export type Extractor = (target: RepoTarget, deps: ExtractorDeps) => Promise } + | { + repoId: string + status: 'partial' + contacts: ScoredContact[] + policies: Partial + } | { repoId: string; status: 'extractor-failed' } diff --git a/services/apps/packages_worker/src/security-contacts/writeContacts.ts b/services/apps/packages_worker/src/security-contacts/writeContacts.ts index 598f009ea3..f96fc88e18 100644 --- a/services/apps/packages_worker/src/security-contacts/writeContacts.ts +++ b/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, + opts: { merge?: boolean } = {}, ): Promise { 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 +type PersistableResult = Extract // 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 { +async function writeContactsChunk(qx: QueryExecutor, chunk: PersistableResult[]): Promise { 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) {