From 26facecef91c9bdd96aa18ff44cdfa8b68097154 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 12:09:57 +0100 Subject: [PATCH 01/13] fix: batch security_contacts insert, drop unsafe per-row writes Signed-off-by: Mouad BANI --- .../src/security-contacts/writeContacts.ts | 48 +++++++++---------- 1 file changed, 22 insertions(+), 26 deletions(-) diff --git a/services/apps/packages_worker/src/security-contacts/writeContacts.ts b/services/apps/packages_worker/src/security-contacts/writeContacts.ts index 6d2196fc53..e5d2b29019 100644 --- a/services/apps/packages_worker/src/security-contacts/writeContacts.ts +++ b/services/apps/packages_worker/src/security-contacts/writeContacts.ts @@ -1,20 +1,18 @@ import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' +import { prepareBulkInsert } from '@crowd/data-access-layer/src/utils' import { RepoPolicies, ScoredContact } from './types' -// 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. +// Advances contacts_last_refreshed without touching contacts/policies, so a failed pass doesn't +// wipe existing data but still isn't reprocessed this sweep. export async function markRepoAttempted(qx: QueryExecutor, repoId: string): Promise { await qx.result('UPDATE repos SET contacts_last_refreshed = NOW() WHERE id = $(repoId)', { repoId, }) } -/** - * 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. - */ +// Assumes at most one writer per repoId at a time (enforced via heartbeat/cancellation in +// processBatch.ts and a non-overlapping schedule in schedule.ts) — no locking here. export async function writeContacts( qx: QueryExecutor, repoId: string, @@ -24,30 +22,28 @@ export async function writeContacts( await qx.tx(async (tx) => { await tx.result('DELETE FROM security_contacts WHERE repo_id = $(repoId)', { repoId }) - for (const c of contacts) { + if (contacts.length > 0) { await tx.result( - `INSERT INTO security_contacts - (repo_id, channel, value, role, name, score, confidence, provenance, last_refreshed) - VALUES - ($(repoId), $(channel), $(value), $(role), $(name), $(score), $(confidence), $(provenance)::jsonb, NOW())`, - { - repoId, - channel: c.channel, - value: c.value, - role: c.role, - name: c.name ?? null, - score: c.score, - confidence: c.confidence, - provenance: JSON.stringify(c.provenance), - }, + prepareBulkInsert( + 'security_contacts', + ['repo_id', 'channel', 'value', 'role', 'name', 'score', 'confidence', 'provenance'], + contacts.map((c) => ({ + repo_id: repoId, + channel: c.channel, + value: c.value, + role: c.role, + name: c.name ?? null, + score: c.score, + confidence: c.confidence, + provenance: JSON.stringify(c.provenance), + })), + ), ) } await tx.result( - // COALESCE preserves previously stored values when a run doesn't (re)discover a field — - // a partial/failed extractor pass must not wipe still-valid policy URLs or the PVR flag. - // vulnerability_reporting_url is PVR-derived, so when PVR is authoritatively resolved we - // overwrite it (clearing it once PVR is disabled); otherwise it is preserved. + // COALESCE preserves a field a partial/failed pass didn't rediscover. vulnerability_reporting_url + // is PVR-derived, so it's overwritten only once PVR is authoritatively resolved. `UPDATE repos SET security_policy_url = COALESCE($(securityPolicyUrl), security_policy_url), vulnerability_reporting_url = CASE WHEN $(pvrResolved) From dc7831355c5f4fc185cb74ea7bab8b447983d9f9 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 12:11:02 +0100 Subject: [PATCH 02/13] fix: bound daily runs to 24h and stop superseded activity retries Signed-off-by: Mouad BANI --- .../src/security-contacts/processBatch.ts | 29 +++++++++-- .../src/security-contacts/schedule.ts | 50 +++++++++++++------ .../src/security-contacts/workflows.ts | 3 ++ 3 files changed, 62 insertions(+), 20 deletions(-) diff --git a/services/apps/packages_worker/src/security-contacts/processBatch.ts b/services/apps/packages_worker/src/security-contacts/processBatch.ts index b01d84a381..5375af0904 100644 --- a/services/apps/packages_worker/src/security-contacts/processBatch.ts +++ b/services/apps/packages_worker/src/security-contacts/processBatch.ts @@ -1,9 +1,13 @@ +import { cancellationSignal, heartbeat } from '@temporalio/activity' + import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' import { getServiceChildLogger } from '@crowd/logging' import { getSecurityContactsConfig } from '../config' +import { parseGithubUrl } from '../enricher/fetchLightRepo' import { mapWithConcurrency } from '../utils/concurrency' +import { fetchRepoTree } from './extractors/gitTree' import { extractPvr } from './extractors/pvr' import { extractManifest } from './extractors/registry' import { extractSecurityContactsFile } from './extractors/securityContactsFile' @@ -114,9 +118,19 @@ function toTarget(row: SweepRow): RepoTarget { async function processRepo( target: RepoTarget, - deps: ExtractorDeps, + baseDeps: Omit, qx: QueryExecutor, ): Promise { + // One tree fetch per repo, shared by extractors that probe well-known paths (see gitTree.ts). + let repoTree: ExtractorDeps['repoTree'] = { paths: null } + try { + const { owner, name } = parseGithubUrl(target.url) + repoTree = await fetchRepoTree(owner, name, baseDeps.githubGet) + } catch { + // not a github.com URL + } + const deps: ExtractorDeps = { ...baseDeps, repoTree } + const results = await Promise.allSettled(EXTRACTORS.map((extract) => extract(target, deps))) // Replace the repo's contacts only when every extractor succeeded. If any failed (transient @@ -158,17 +172,20 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise = { fetchTimeoutMs: FETCH_TIMEOUT_MS, userAgent: config.userAgent, githubGet: (path, opts) => githubApiGet(path, FETCH_TIMEOUT_MS, opts), } const targets = batch.map(toTarget) - // 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. + // mapWithConcurrency is fail-fast: a per-repo DB error is caught below so it doesn't abort the + // batch, but a cancelled task (superseded by a newer activity attempt, see workflows.ts) is + // left to throw so it stops scheduling further repos instead of racing the new attempt. await mapWithConcurrency(targets, CONCURRENCY, async (target) => { + if (cancellationSignal().aborted) { + throw new Error('Security contacts batch cancelled — superseded by a newer activity attempt') + } try { await processRepo(target, deps, qx) } catch (err) { @@ -176,6 +193,8 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise 0 forever). await markRepoAttempted(qx, target.repoId).catch(() => undefined) + } finally { + heartbeat() } }) diff --git a/services/apps/packages_worker/src/security-contacts/schedule.ts b/services/apps/packages_worker/src/security-contacts/schedule.ts index 1ea3a4ec5b..ebd2762d86 100644 --- a/services/apps/packages_worker/src/security-contacts/schedule.ts +++ b/services/apps/packages_worker/src/security-contacts/schedule.ts @@ -3,13 +3,34 @@ import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/clien import { svc } from '../service' import { ingestSecurityContacts } from '../workflows' +// Unlike workflowRunTimeout, workflowExecutionTimeout bounds the entire continueAsNew chain — +// caps a run to 24h so it can't still be going when the next day's tick fires. +const SCHEDULE_ID = 'security-contacts-ingestion' +const WORKFLOW_EXECUTION_TIMEOUT = '24 hours' + +function scheduleAction() { + return { + type: 'startWorkflow' as const, + workflowType: ingestSecurityContacts, + workflowId: 'security-contacts-daily', + taskQueue: 'packages-worker', + workflowExecutionTimeout: WORKFLOW_EXECUTION_TIMEOUT, + retry: { + initialInterval: '30 seconds', + backoffCoefficient: 2, + maximumAttempts: 3, + }, + args: [] as [], + } +} + export async function scheduleSecurityContactsIngestion(): Promise { const { temporal } = svc if (!temporal) throw new Error('Temporal client not initialized') try { await temporal.schedule.create({ - scheduleId: 'security-contacts-ingestion', + scheduleId: SCHEDULE_ID, spec: { cronExpressions: ['0 6 * * *'], }, @@ -17,23 +38,22 @@ export async function scheduleSecurityContactsIngestion(): Promise { overlap: ScheduleOverlapPolicy.SKIP, catchupWindow: '1 hour', }, - action: { - type: 'startWorkflow', - workflowType: ingestSecurityContacts, - workflowId: 'security-contacts-daily', - taskQueue: 'packages-worker', - workflowRunTimeout: '24 hours', - retry: { - initialInterval: '30 seconds', - backoffCoefficient: 2, - maximumAttempts: 3, - }, - args: [], - }, + action: scheduleAction(), }) } catch (err) { if (err instanceof ScheduleAlreadyRunning) { - svc.log.info('Schedule security-contacts-ingestion already exists, skipping creation.') + // create() only runs once; reconcile on every startup so action/policy changes reach + // the already-existing live schedule. + svc.log.info('Schedule security-contacts-ingestion already exists, reconciling action.') + const handle = temporal.schedule.getHandle(SCHEDULE_ID) + await handle.update((prev) => ({ + ...prev, + policies: { + ...prev.policies, + overlap: ScheduleOverlapPolicy.SKIP, + }, + action: scheduleAction(), + })) } else { throw err } diff --git a/services/apps/packages_worker/src/security-contacts/workflows.ts b/services/apps/packages_worker/src/security-contacts/workflows.ts index 2b15660963..fa9c64b072 100644 --- a/services/apps/packages_worker/src/security-contacts/workflows.ts +++ b/services/apps/packages_worker/src/security-contacts/workflows.ts @@ -4,6 +4,9 @@ import type * as activities from './activities' const acts = proxyActivities({ startToCloseTimeout: '30 minutes', + // Lets a stale attempt (see processBatch.ts heartbeat) detect it's been superseded by a retry + // and stop, instead of both running concurrently against the same repos. + heartbeatTimeout: '2 minutes', retry: { initialInterval: '30 seconds', backoffCoefficient: 2, From f4405d76eef9be91cf1f24f630d3004f78aa1800 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 12:18:14 +0100 Subject: [PATCH 03/13] fix: treat HTTP 451 as a permanently-absent status Signed-off-by: Mouad BANI --- .../packages_worker/src/security-contacts/extractors/http.ts | 3 ++- .../apps/packages_worker/src/security-contacts/githubToken.ts | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/services/apps/packages_worker/src/security-contacts/extractors/http.ts b/services/apps/packages_worker/src/security-contacts/extractors/http.ts index 8f1c625e20..86dd969502 100644 --- a/services/apps/packages_worker/src/security-contacts/extractors/http.ts +++ b/services/apps/packages_worker/src/security-contacts/extractors/http.ts @@ -8,7 +8,8 @@ export function registryHeaders(userAgent: string): Record { // failures (5xx/...) are treated as failures and the pipeline preserves data instead of // wiping it. 422 is included because GitHub's PVR endpoint returns it (per-repo, non-transient) // when the flag can't be determined — that must read as "unknown", not block the whole repo. -const ABSENT_STATUSES = new Set([404, 410, 422]) +// 451 (Unavailable For Legal Reasons) is likewise permanent. +const ABSENT_STATUSES = new Set([404, 410, 422, 451]) // Registry rate-limit / overload responses: retried in-process (honoring Retry-After) so a brief // throttle doesn't fail the extractor and cost the repo a whole refresh cadence. diff --git a/services/apps/packages_worker/src/security-contacts/githubToken.ts b/services/apps/packages_worker/src/security-contacts/githubToken.ts index 7c5cbc38da..a7d9022147 100644 --- a/services/apps/packages_worker/src/security-contacts/githubToken.ts +++ b/services/apps/packages_worker/src/security-contacts/githubToken.ts @@ -17,7 +17,8 @@ const GITHUB_API = 'https://api.github.com' // Genuinely-absent / not-determinable → null body (see http.ts for the same set). 422 covers the // PVR endpoint returning "can't determine" per-repo; it must read as unknown, not a hard failure. -const ABSENT_STATUSES = new Set([404, 410, 422]) +// 451 (Unavailable For Legal Reasons) is likewise permanent. +const ABSENT_STATUSES = new Set([404, 410, 422, 451]) // App-wide ceiling on concurrent GitHub requests. GitHub's secondary limit rejects bursts of // >100 concurrent requests from one app; staying under that (across all installations) is the From 647207ba3389647ce5b75eab6efce3c88de1d6d9 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 12:18:57 +0100 Subject: [PATCH 04/13] perf: skip known-absent Contents probes via one git-tree call Signed-off-by: Mouad BANI --- .../security-contacts/extractors/gitTree.ts | 39 +++++++++++++++++++ .../extractors/securityContactsFile.ts | 2 + .../extractors/securityInsights.ts | 2 + .../extractors/securityMd.ts | 2 + .../src/security-contacts/types.ts | 2 + 5 files changed, 47 insertions(+) create mode 100644 services/apps/packages_worker/src/security-contacts/extractors/gitTree.ts diff --git a/services/apps/packages_worker/src/security-contacts/extractors/gitTree.ts b/services/apps/packages_worker/src/security-contacts/extractors/gitTree.ts new file mode 100644 index 0000000000..526c54a230 --- /dev/null +++ b/services/apps/packages_worker/src/security-contacts/extractors/gitTree.ts @@ -0,0 +1,39 @@ +import { ExtractorDeps } from '../types' + +/** null means unresolved (empty repo, truncated, fetch failure) — callers must fall back to probing. */ +export interface RepoTree { + paths: Set | null +} + +interface GitTreeEntry { + path?: unknown + type?: unknown +} + +interface GitTreeResponse { + tree?: GitTreeEntry[] + truncated?: boolean +} + +// `HEAD` resolves to the default branch on the git data API. +export async function fetchRepoTree( + owner: string, + name: string, + githubGet: ExtractorDeps['githubGet'], +): Promise { + try { + const { text } = await githubGet(`/repos/${owner}/${name}/git/trees/HEAD?recursive=1`) + if (!text) return { paths: null } + + const doc = JSON.parse(text) as GitTreeResponse + if (doc.truncated || !Array.isArray(doc.tree)) return { paths: null } + + const paths = new Set() + for (const entry of doc.tree) { + if (entry.type === 'blob' && typeof entry.path === 'string') paths.add(entry.path) + } + return { paths } + } catch { + return { paths: null } + } +} diff --git a/services/apps/packages_worker/src/security-contacts/extractors/securityContactsFile.ts b/services/apps/packages_worker/src/security-contacts/extractors/securityContactsFile.ts index 3807b14aab..ad29561f6d 100644 --- a/services/apps/packages_worker/src/security-contacts/extractors/securityContactsFile.ts +++ b/services/apps/packages_worker/src/security-contacts/extractors/securityContactsFile.ts @@ -55,6 +55,8 @@ export const extractSecurityContactsFile: Extractor = async (target, deps) => { return { contacts: [], policies: {} } } + if (deps.repoTree.paths && !deps.repoTree.paths.has(PATH)) return { contacts: [], policies: {} } + const { text } = await deps.githubGet(`/repos/${owner}/${name}/contents/${PATH}`, { raw: true }) if (!text) return { contacts: [], policies: {} } diff --git a/services/apps/packages_worker/src/security-contacts/extractors/securityInsights.ts b/services/apps/packages_worker/src/security-contacts/extractors/securityInsights.ts index 86ebf4abb0..985591214c 100644 --- a/services/apps/packages_worker/src/security-contacts/extractors/securityInsights.ts +++ b/services/apps/packages_worker/src/security-contacts/extractors/securityInsights.ts @@ -211,8 +211,10 @@ export const extractSecurityInsights: Extractor = async (target, deps) => { } const fetchedAt = new Date().toISOString() + const { paths: treePaths } = deps.repoTree for (const path of PATHS) { + if (treePaths && !treePaths.has(path)) continue const { text } = await deps.githubGet(`/repos/${owner}/${name}/contents/${path}`, { raw: true }) if (!text) continue diff --git a/services/apps/packages_worker/src/security-contacts/extractors/securityMd.ts b/services/apps/packages_worker/src/security-contacts/extractors/securityMd.ts index 89d7e7d3ef..9105e891de 100644 --- a/services/apps/packages_worker/src/security-contacts/extractors/securityMd.ts +++ b/services/apps/packages_worker/src/security-contacts/extractors/securityMd.ts @@ -109,7 +109,9 @@ export const extractSecurityMd: Extractor = async (target, deps) => { } const fetchedAt = new Date().toISOString() + const { paths: treePaths } = deps.repoTree for (const path of PATHS) { + if (treePaths && !treePaths.has(path)) continue const { text } = await deps.githubGet(`/repos/${owner}/${name}/contents/${path}`, { raw: true }) if (text) return parseSecurityMd(text, owner, name, path, fetchedAt) } diff --git a/services/apps/packages_worker/src/security-contacts/types.ts b/services/apps/packages_worker/src/security-contacts/types.ts index b3cd480732..207aff4df4 100644 --- a/services/apps/packages_worker/src/security-contacts/types.ts +++ b/services/apps/packages_worker/src/security-contacts/types.ts @@ -79,6 +79,8 @@ export interface ExtractorDeps { * and 429/secondary-limit backoff internally. `raw` selects the raw media type (file contents). */ githubGet: (path: string, opts?: { raw?: boolean }) => Promise + /** Default-branch file paths from one git-tree call; `null` means unresolved — probe as before. */ + repoTree: { paths: Set | null } } export type Extractor = (target: RepoTarget, deps: ExtractorDeps) => Promise From 829220cba6ce8be21c31d04613daf9b9d3458491 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 14:42:22 +0100 Subject: [PATCH 05/13] fix: soft-delete security_contacts on refresh instead of hard delete Signed-off-by: Mouad BANI --- ...3036800__security_contacts_soft_delete.sql | 1 + .../src/security-contacts/writeContacts.ts | 19 ++++++++++++++++++- 2 files changed, 19 insertions(+), 1 deletion(-) create mode 100644 backend/src/osspckgs/migrations/V1783036800__security_contacts_soft_delete.sql diff --git a/backend/src/osspckgs/migrations/V1783036800__security_contacts_soft_delete.sql b/backend/src/osspckgs/migrations/V1783036800__security_contacts_soft_delete.sql new file mode 100644 index 0000000000..3bfd8ff343 --- /dev/null +++ b/backend/src/osspckgs/migrations/V1783036800__security_contacts_soft_delete.sql @@ -0,0 +1 @@ +ALTER TABLE security_contacts ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ; diff --git a/services/apps/packages_worker/src/security-contacts/writeContacts.ts b/services/apps/packages_worker/src/security-contacts/writeContacts.ts index e5d2b29019..315dd2f558 100644 --- a/services/apps/packages_worker/src/security-contacts/writeContacts.ts +++ b/services/apps/packages_worker/src/security-contacts/writeContacts.ts @@ -13,6 +13,11 @@ export async function markRepoAttempted(qx: QueryExecutor, repoId: string): Prom // Assumes at most one writer per repoId at a time (enforced via heartbeat/cancellation in // processBatch.ts and a non-overlapping schedule in schedule.ts) — no locking here. +// +// Soft-delete: mark every currently-active row stale, then upsert this pass's contacts, which +// revives (clears deleted_at on) whatever was rediscovered. Anything not rediscovered stays +// soft-deleted instead of being destroyed, preserving history. Readers of this table must filter +// on deleted_at IS NULL. export async function writeContacts( qx: QueryExecutor, repoId: string, @@ -20,7 +25,10 @@ export async function writeContacts( policies: Partial, ): Promise { await qx.tx(async (tx) => { - await tx.result('DELETE FROM security_contacts WHERE repo_id = $(repoId)', { repoId }) + 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( @@ -37,6 +45,15 @@ export async function writeContacts( confidence: c.confidence, provenance: JSON.stringify(c.provenance), })), + `(repo_id, channel, value) DO UPDATE SET + role = EXCLUDED.role, + name = EXCLUDED.name, + score = EXCLUDED.score, + confidence = EXCLUDED.confidence, + provenance = EXCLUDED.provenance, + last_refreshed = NOW(), + updated_at = NOW(), + deleted_at = NULL`, ), ) } From 1633ec28db8a91337b4ef6e250b4126efeea542a Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 14:56:11 +0100 Subject: [PATCH 06/13] fix: exclude soft-deleted records on queries Signed-off-by: Mouad BANI --- .../packages_worker/src/security-contacts/processBatch.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/services/apps/packages_worker/src/security-contacts/processBatch.ts b/services/apps/packages_worker/src/security-contacts/processBatch.ts index 5375af0904..4718fded0a 100644 --- a/services/apps/packages_worker/src/security-contacts/processBatch.ts +++ b/services/apps/packages_worker/src/security-contacts/processBatch.ts @@ -85,12 +85,12 @@ async function fetchBatch(qx: QueryExecutor): Promise { r.contacts_last_refreshed IS NULL -- evaluated but no contacts found yet → retry on the daily cadence OR ( - NOT EXISTS (SELECT 1 FROM security_contacts sc WHERE sc.repo_id = r.id) + NOT EXISTS (SELECT 1 FROM security_contacts sc WHERE sc.repo_id = r.id AND sc.deleted_at IS NULL) AND r.contacts_last_refreshed < NOW() - INTERVAL '$(dailyIntervalHours) hours' ) -- already enriched (has contacts) → refresh on the weekly cadence OR ( - EXISTS (SELECT 1 FROM security_contacts sc WHERE sc.repo_id = r.id) + EXISTS (SELECT 1 FROM security_contacts sc WHERE sc.repo_id = r.id AND sc.deleted_at IS NULL) AND r.contacts_last_refreshed < NOW() - INTERVAL '$(weeklyIntervalHours) hours' ) ) From 04aad4750208ad51e070e2aa32a1262a97970b69 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 15:14:30 +0100 Subject: [PATCH 07/13] fix: workflow hearbeat Signed-off-by: Mouad BANI --- .../src/security-contacts/processBatch.ts | 43 +++++++++++-------- 1 file changed, 25 insertions(+), 18 deletions(-) diff --git a/services/apps/packages_worker/src/security-contacts/processBatch.ts b/services/apps/packages_worker/src/security-contacts/processBatch.ts index 4718fded0a..a17c90e861 100644 --- a/services/apps/packages_worker/src/security-contacts/processBatch.ts +++ b/services/apps/packages_worker/src/security-contacts/processBatch.ts @@ -179,24 +179,31 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise { - if (cancellationSignal().aborted) { - throw new Error('Security contacts batch cancelled — superseded by a newer activity attempt') - } - try { - await processRepo(target, deps, qx) - } catch (err) { - log.error({ repoId: target.repoId, errMsg: (err as Error).message }, 'Repo processing failed') - // Best-effort mark so a persistently-failing repo drains on cadence rather than making the - // sweep hot-loop (it would otherwise stay eligible and keep processed > 0 forever). - await markRepoAttempted(qx, target.repoId).catch(() => undefined) - } finally { - heartbeat() - } - }) + // Heartbeat on a fixed cadence rather than per-repo completion: a single repo's extractors + // (sequential per-package registry fetches, cargo throttling) can outlast the 2-minute + // heartbeatTimeout even while every CONCURRENCY slot is still busy, so relying on task + // completions to drive the heartbeat can miss the deadline. + const heartbeatTimer = setInterval(() => heartbeat(), 30_000) + try { + // mapWithConcurrency is fail-fast: a per-repo DB error is caught below so it doesn't abort + // the batch, but a cancelled task (superseded by a newer activity attempt, see workflows.ts) + // is left to throw so it stops scheduling further repos instead of racing the new attempt. + await mapWithConcurrency(targets, CONCURRENCY, async (target) => { + if (cancellationSignal().aborted) { + throw new Error('Security contacts batch cancelled — superseded by a newer activity attempt') + } + try { + await processRepo(target, deps, qx) + } catch (err) { + log.error({ repoId: target.repoId, errMsg: (err as Error).message }, 'Repo processing failed') + // Best-effort mark so a persistently-failing repo drains on cadence rather than making the + // sweep hot-loop (it would otherwise stay eligible and keep processed > 0 forever). + await markRepoAttempted(qx, target.repoId).catch(() => undefined) + } + }) + } finally { + clearInterval(heartbeatTimer) + } log.info({ processed: targets.length }, 'Security contacts batch complete') return { processed: targets.length } From 33d8376c02505498c9696bda595a028b58ac851a Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 15:19:24 +0100 Subject: [PATCH 08/13] chore: remove verbose comments Signed-off-by: Mouad BANI --- .../src/security-contacts/extractors/http.ts | 10 ++--- .../src/security-contacts/githubToken.ts | 21 +++------- .../src/security-contacts/processBatch.ts | 40 ++++++------------- .../src/security-contacts/types.ts | 19 +++------ .../src/security-contacts/writeContacts.ts | 15 +++---- 5 files changed, 32 insertions(+), 73 deletions(-) diff --git a/services/apps/packages_worker/src/security-contacts/extractors/http.ts b/services/apps/packages_worker/src/security-contacts/extractors/http.ts index 86dd969502..4753445c2b 100644 --- a/services/apps/packages_worker/src/security-contacts/extractors/http.ts +++ b/services/apps/packages_worker/src/security-contacts/extractors/http.ts @@ -4,15 +4,11 @@ export function registryHeaders(userAgent: string): Record { return { 'User-Agent': userAgent } } -// Genuinely-absent / not-determinable → null body; every other non-200 throws so transient -// failures (5xx/...) are treated as failures and the pipeline preserves data instead of -// wiping it. 422 is included because GitHub's PVR endpoint returns it (per-repo, non-transient) -// when the flag can't be determined — that must read as "unknown", not block the whole repo. -// 451 (Unavailable For Legal Reasons) is likewise permanent. +// Genuinely-absent/not-determinable → null body; every other non-200 throws so the pipeline +// preserves existing data instead of wiping it. 422 covers GitHub's PVR "can't determine". const ABSENT_STATUSES = new Set([404, 410, 422, 451]) -// Registry rate-limit / overload responses: retried in-process (honoring Retry-After) so a brief -// throttle doesn't fail the extractor and cost the repo a whole refresh cadence. +// Retried in-process (honoring Retry-After) so a brief throttle doesn't cost a whole cadence. const RATE_LIMIT_STATUSES = new Set([429, 503]) const MAX_RATE_LIMIT_RETRIES = 3 diff --git a/services/apps/packages_worker/src/security-contacts/githubToken.ts b/services/apps/packages_worker/src/security-contacts/githubToken.ts index a7d9022147..c118269e96 100644 --- a/services/apps/packages_worker/src/security-contacts/githubToken.ts +++ b/services/apps/packages_worker/src/security-contacts/githubToken.ts @@ -15,18 +15,13 @@ const log = getServiceChildLogger('security-contacts:github-token') const GITHUB_API = 'https://api.github.com' -// Genuinely-absent / not-determinable → null body (see http.ts for the same set). 422 covers the -// PVR endpoint returning "can't determine" per-repo; it must read as unknown, not a hard failure. -// 451 (Unavailable For Legal Reasons) is likewise permanent. +// Genuinely-absent / not-determinable → null body, not a failure (see http.ts for the same set). +// 422 covers GitHub's PVR endpoint returning "can't determine" per-repo; 451 is permanent removal. const ABSENT_STATUSES = new Set([404, 410, 422, 451]) -// App-wide ceiling on concurrent GitHub requests. GitHub's secondary limit rejects bursts of -// >100 concurrent requests from one app; staying under that (across all installations) is the -// single most effective guard against secondary-limit 429s at high repo concurrency. +// App-wide ceiling on concurrent requests — GitHub's secondary limit rejects bursts >100/app. const MAX_CONCURRENT_GITHUB_REQUESTS = 50 -// Bound the park/switch/backoff retry loop so a persistently-limited request eventually surfaces -// as a failure (which the pipeline treats as transient and preserves existing data). const MAX_RATE_LIMIT_RETRIES = 6 /** Minimal async semaphore with fair FIFO hand-off, used to cap concurrent GitHub requests. */ @@ -127,13 +122,9 @@ async function fetchOnce( } /** - * Rate-limit-safe GitHub API GET. Selects an installation from the pool, sleeps if all are parked, - * feeds response budget headers back so exhausted installations get parked before they 403, and on - * a rate-limit response parks (primary) or waits out Retry-After (secondary, app-wide) then retries - * on another installation. Falls back to a single unauthenticated request when no App is configured. - * - * Returns text on 200; null body for absent resources (404/410/422); throws on other non-200s and - * once the retry budget is exhausted (callers treat throws as transient and preserve existing data). + * Rate-limit-safe GitHub API GET. Rotates across installations in the pool, parking exhausted + * or rate-limited ones and retrying, up to MAX_RATE_LIMIT_RETRIES. Falls back to a single + * unauthenticated request when no App is configured. */ export async function githubApiGet( path: string, diff --git a/services/apps/packages_worker/src/security-contacts/processBatch.ts b/services/apps/packages_worker/src/security-contacts/processBatch.ts index a17c90e861..b2e9c5fd45 100644 --- a/services/apps/packages_worker/src/security-contacts/processBatch.ts +++ b/services/apps/packages_worker/src/security-contacts/processBatch.ts @@ -35,18 +35,11 @@ export interface BatchResult { processed: number } -// Two-tier refresh cadence (the worker runs on a daily cron). Just under 24h/168h so a repo -// processed at ~06:00 is eligible again at the next daily/weekly tick rather than slipping a day. -const DAILY_INTERVAL_HOURS = 20 // repos never evaluated or with no contacts yet -const WEEKLY_INTERVAL_HOURS = 156 // already-enriched repos (have contacts) - -// Tuned for throughput within the platform ceilings: -// - CONCURRENCY: parallel repos. GitHub calls go through the authed Contents API via a -// rate-limit-aware pool (per-installation budget parking + app-wide concurrency gate + -// Retry-After backoff in githubToken), so high repo concurrency won't trip GitHub limits. -// - FETCH_TIMEOUT_MS: generous enough for slow registries (Maven metadata/POM) without hanging slots. -// - BATCH_SIZE: bounded by the 30-min activity timeout (worst case: an all-cargo batch throttled -// to crates.io's 1 req/s finishes ~8 min). +// Just under 24h/168h so a repo processed at ~06:00 is still eligible at the next tick. +const DAILY_INTERVAL_HOURS = 20 // no contacts found yet +const WEEKLY_INTERVAL_HOURS = 156 // already has contacts + +// GitHub calls are rate-limit-aware (see githubToken.ts), so high repo concurrency is safe. const CONCURRENCY = 100 const FETCH_TIMEOUT_MS = 15000 const BATCH_SIZE = 500 @@ -121,7 +114,7 @@ async function processRepo( baseDeps: Omit, qx: QueryExecutor, ): Promise { - // One tree fetch per repo, shared by extractors that probe well-known paths (see gitTree.ts). + // One tree fetch per repo, shared by extractors that probe well-known paths. let repoTree: ExtractorDeps['repoTree'] = { paths: null } try { const { owner, name } = parseGithubUrl(target.url) @@ -133,9 +126,7 @@ async function processRepo( const results = await Promise.allSettled(EXTRACTORS.map((extract) => extract(target, deps))) - // 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. + // 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) { log.warn( @@ -158,8 +149,7 @@ async function processRepo( } } - // A2 veto: B1 may emit a github-pvr contact from redirect language; drop it when A2 - // authoritatively reports PVR disabled (Option C from the design discussion). + // A2 veto: drop B1's github-pvr guess when A2 authoritatively reports PVR disabled. if (policies.pvrEnabled === false) { contacts = contacts.filter((c) => c.channel !== 'github-pvr') } @@ -179,15 +169,12 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise heartbeat(), 30_000) try { - // mapWithConcurrency is fail-fast: a per-repo DB error is caught below so it doesn't abort - // the batch, but a cancelled task (superseded by a newer activity attempt, see workflows.ts) - // is left to throw so it stops scheduling further repos instead of racing the new attempt. + // A cancelled task (superseded by a newer activity attempt) is left to throw so it stops + // scheduling further repos instead of racing the new attempt. await mapWithConcurrency(targets, CONCURRENCY, async (target) => { if (cancellationSignal().aborted) { throw new Error('Security contacts batch cancelled — superseded by a newer activity attempt') @@ -196,8 +183,7 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise 0 forever). + // Best-effort: keeps a persistently-failing repo from hot-looping the sweep. await markRepoAttempted(qx, target.repoId).catch(() => undefined) } }) diff --git a/services/apps/packages_worker/src/security-contacts/types.ts b/services/apps/packages_worker/src/security-contacts/types.ts index 207aff4df4..aad43d4155 100644 --- a/services/apps/packages_worker/src/security-contacts/types.ts +++ b/services/apps/packages_worker/src/security-contacts/types.ts @@ -8,24 +8,20 @@ export type SourceTier = 'A' | 'B' | 'C' | 'D' /** Where a single contact value was observed, for auditability and corroboration scoring. */ export interface ProvenanceEntry { - /** Human-readable source identifier, e.g. 'security-insights', 'pvr', 'security.md'. */ source: string sourceTier: SourceTier - /** Path or URL the value was read from, when applicable. */ path?: string - /** When this worker fetched the source. ISO-8601. */ fetchedAt: string - /** When the source itself declared the value (e.g. SECURITY-INSIGHTS last-updated). ISO-8601. */ + /** When the source itself declared the value, if different from fetchedAt. ISO-8601. */ declaredAt?: string } -/** Extractor output, before reconciliation and scoring. */ export interface RawContact { channel: ContactChannel value: string role: ContactRole name?: string - /** Username an A3 email was resolved from — used only to identity-link a bare github-handle. */ + /** Username an email was resolved from — used to identity-link a bare github-handle. */ handle?: string tier: SourceTier provenance: ProvenanceEntry[] @@ -63,23 +59,18 @@ export interface RepoTarget { packages: RepoPackage[] } -/** Result of a GitHub API GET routed through the rate-limit-aware pool. */ export interface GithubGetResult { status: number - /** Response body (raw file text or JSON string); null for absent resources (404/410/422). */ + /** Null for absent resources (404/410/422/451). */ text: string | null } export interface ExtractorDeps { fetchTimeoutMs: number - /** Sent on registry calls; required (crates.io rejects requests without an identifying UA). */ + /** Required — crates.io rejects requests without an identifying UA. */ userAgent: string - /** - * Pool-aware, rate-limit-safe GitHub API GET. Handles installation selection, budget parking, - * and 429/secondary-limit backoff internally. `raw` selects the raw media type (file contents). - */ githubGet: (path: string, opts?: { raw?: boolean }) => Promise - /** Default-branch file paths from one git-tree call; `null` means unresolved — probe as before. */ + /** Default-branch file paths from one git-tree call; null means unresolved — probe as before. */ repoTree: { paths: Set | null } } diff --git a/services/apps/packages_worker/src/security-contacts/writeContacts.ts b/services/apps/packages_worker/src/security-contacts/writeContacts.ts index 315dd2f558..6a6e4eecdc 100644 --- a/services/apps/packages_worker/src/security-contacts/writeContacts.ts +++ b/services/apps/packages_worker/src/security-contacts/writeContacts.ts @@ -3,21 +3,17 @@ import { prepareBulkInsert } from '@crowd/data-access-layer/src/utils' import { RepoPolicies, ScoredContact } from './types' -// Advances contacts_last_refreshed without touching contacts/policies, so a failed pass doesn't -// wipe existing data but still isn't reprocessed this sweep. +// Advances contacts_last_refreshed only, so a failed pass isn't reprocessed this sweep. export async function markRepoAttempted(qx: QueryExecutor, repoId: string): Promise { await qx.result('UPDATE repos SET contacts_last_refreshed = NOW() WHERE id = $(repoId)', { repoId, }) } -// Assumes at most one writer per repoId at a time (enforced via heartbeat/cancellation in -// processBatch.ts and a non-overlapping schedule in schedule.ts) — no locking here. +// Assumes at most one writer per repoId at a time (see processBatch.ts/schedule.ts) — no locking. // -// Soft-delete: mark every currently-active row stale, then upsert this pass's contacts, which -// revives (clears deleted_at on) whatever was rediscovered. Anything not rediscovered stays -// soft-deleted instead of being destroyed, preserving history. Readers of this table must filter -// on deleted_at IS NULL. +// 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. export async function writeContacts( qx: QueryExecutor, repoId: string, @@ -59,8 +55,7 @@ export async function writeContacts( } await tx.result( - // COALESCE preserves a field a partial/failed pass didn't rediscover. vulnerability_reporting_url - // is PVR-derived, so it's overwritten only once PVR is authoritatively resolved. + // vulnerability_reporting_url is overwritten only once PVR is authoritatively resolved. `UPDATE repos SET security_policy_url = COALESCE($(securityPolicyUrl), security_policy_url), vulnerability_reporting_url = CASE WHEN $(pvrResolved) From 8e4a6ea3db008fb87276b7ecdbb462e32446b9b3 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 15:27:11 +0100 Subject: [PATCH 09/13] fix: wrap heartbeat in try/catch Signed-off-by: Mouad BANI --- .../packages_worker/src/security-contacts/processBatch.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/services/apps/packages_worker/src/security-contacts/processBatch.ts b/services/apps/packages_worker/src/security-contacts/processBatch.ts index b2e9c5fd45..c2ff87eb09 100644 --- a/services/apps/packages_worker/src/security-contacts/processBatch.ts +++ b/services/apps/packages_worker/src/security-contacts/processBatch.ts @@ -171,7 +171,13 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise heartbeat(), 30_000) + const heartbeatTimer = setInterval(() => { + try { + heartbeat() + } catch (err) { + log.warn({ errMsg: (err as Error).message }, 'Heartbeat failed') + } + }, 30_000) try { // A cancelled task (superseded by a newer activity attempt) is left to throw so it stops // scheduling further repos instead of racing the new attempt. From b746e41aab8da5c3de8a349c440cf80d5fa6ff4d Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 15:28:32 +0100 Subject: [PATCH 10/13] chore: lint Signed-off-by: Mouad BANI --- .../src/security-contacts/processBatch.ts | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/services/apps/packages_worker/src/security-contacts/processBatch.ts b/services/apps/packages_worker/src/security-contacts/processBatch.ts index c2ff87eb09..9033a8dfa5 100644 --- a/services/apps/packages_worker/src/security-contacts/processBatch.ts +++ b/services/apps/packages_worker/src/security-contacts/processBatch.ts @@ -183,12 +183,17 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise { if (cancellationSignal().aborted) { - throw new Error('Security contacts batch cancelled — superseded by a newer activity attempt') + throw new Error( + 'Security contacts batch cancelled — superseded by a newer activity attempt', + ) } try { await processRepo(target, deps, qx) } catch (err) { - log.error({ repoId: target.repoId, errMsg: (err as Error).message }, 'Repo processing failed') + log.error( + { repoId: target.repoId, errMsg: (err as Error).message }, + 'Repo processing failed', + ) // Best-effort: keeps a persistently-failing repo from hot-looping the sweep. await markRepoAttempted(qx, target.repoId).catch(() => undefined) } From 36ecfbdfc0b174b2e41597e1d8064ce3cbcac96f Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 15:38:26 +0100 Subject: [PATCH 11/13] perf: add partial indexe on security_contacts Signed-off-by: Mouad BANI --- .../migrations/V1783036800__security_contacts_soft_delete.sql | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/backend/src/osspckgs/migrations/V1783036800__security_contacts_soft_delete.sql b/backend/src/osspckgs/migrations/V1783036800__security_contacts_soft_delete.sql index 3bfd8ff343..84b9a26a23 100644 --- a/backend/src/osspckgs/migrations/V1783036800__security_contacts_soft_delete.sql +++ b/backend/src/osspckgs/migrations/V1783036800__security_contacts_soft_delete.sql @@ -1 +1,5 @@ ALTER TABLE security_contacts ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ; + +CREATE INDEX IF NOT EXISTS security_contacts_repo_active_idx + ON security_contacts (repo_id) + WHERE deleted_at IS NULL; \ No newline at end of file From 8f3669f46bd201cfba92a32337c3a477b77d2b9e Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 16:46:28 +0100 Subject: [PATCH 12/13] feat: filter out junk contact records Signed-off-by: Mouad BANI --- .../src/security-contacts/reconcile.ts | 31 ++++++++++++++++++- 1 file changed, 30 insertions(+), 1 deletion(-) diff --git a/services/apps/packages_worker/src/security-contacts/reconcile.ts b/services/apps/packages_worker/src/security-contacts/reconcile.ts index 545c343007..7fc2eb6d24 100644 --- a/services/apps/packages_worker/src/security-contacts/reconcile.ts +++ b/services/apps/packages_worker/src/security-contacts/reconcile.ts @@ -20,6 +20,35 @@ const ROLE_PRIORITY: Record = { const TIER_RANK: Record = { A: 4, B: 3, C: 2, D: 1 } +// RFC 2606 reserved domains — always unedited template placeholders (e.g. security@example.com). +const PLACEHOLDER_EMAIL_DOMAINS = new Set([ + 'example.com', + 'example.org', + 'example.net', + 'example.edu', +]) + +// Generic GitHub/Dependabot help pages that templated SECURITY.md files link to — never a +// project-specific contact, unlike an actual github.com///... URL. +const GENERIC_URL_HOSTS = new Set(['docs.github.com', 'dependabot.com', 'www.dependabot.com']) + +function isJunkContact(c: RawContact): boolean { + if (c.channel === 'email') { + const domain = c.value.split('@')[1]?.toLowerCase().trim() + return domain != null && PLACEHOLDER_EMAIL_DOMAINS.has(domain) + } + if (c.channel === 'url' || c.channel === 'web-form') { + let host: string + try { + host = new URL(c.value).hostname.toLowerCase() + } catch { + return true + } + return GENERIC_URL_HOSTS.has(host) || host === 'localhost' || host.startsWith('127.') + } + return false +} + function normalizeValue(channel: ContactChannel, value: string): string { const v = value.trim() return channel === 'email' || channel === 'github-handle' ? v.toLowerCase() : v @@ -90,7 +119,7 @@ function identityLinkMerge(contacts: RawContact[]): RawContact[] { } export function reconcile(contacts: RawContact[], now: Date = new Date()): ScoredContact[] { - const merged = identityLinkMerge(exactMatchMerge(contacts)) + const merged = identityLinkMerge(exactMatchMerge(contacts.filter((c) => !isJunkContact(c)))) const scored: ScoredContact[] = merged.map((c) => { const contact = { ...c, provenance: dedupeProvenance(c.provenance) } From a958c144e9445548ba80a0167202af3603da44e0 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Thu, 2 Jul 2026 17:02:49 +0100 Subject: [PATCH 13/13] fix: isJustnContact url validation Signed-off-by: Mouad BANI --- .../src/security-contacts/reconcile.ts | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/services/apps/packages_worker/src/security-contacts/reconcile.ts b/services/apps/packages_worker/src/security-contacts/reconcile.ts index 7fc2eb6d24..40d4f46f22 100644 --- a/services/apps/packages_worker/src/security-contacts/reconcile.ts +++ b/services/apps/packages_worker/src/security-contacts/reconcile.ts @@ -32,18 +32,27 @@ const PLACEHOLDER_EMAIL_DOMAINS = new Set([ // project-specific contact, unlike an actual github.com///... URL. const GENERIC_URL_HOSTS = new Set(['docs.github.com', 'dependabot.com', 'www.dependabot.com']) +const HAS_SCHEME_RE = /^[a-z][a-z0-9+.-]*:\/\//i + +// Some upstream sources (e.g. SECURITY-INSIGHTS {type:"url"} entries) don't enforce a scheme — +// only add one to resolve the host, never mutate the stored contact value. +function urlHost(value: string): string | null { + const candidate = HAS_SCHEME_RE.test(value) ? value : `https://${value}` + try { + return new URL(candidate).hostname.toLowerCase() + } catch { + return null + } +} + function isJunkContact(c: RawContact): boolean { if (c.channel === 'email') { const domain = c.value.split('@')[1]?.toLowerCase().trim() return domain != null && PLACEHOLDER_EMAIL_DOMAINS.has(domain) } if (c.channel === 'url' || c.channel === 'web-form') { - let host: string - try { - host = new URL(c.value).hostname.toLowerCase() - } catch { - return true - } + const host = urlHost(c.value) + if (host == null) return true return GENERIC_URL_HOSTS.has(host) || host === 'localhost' || host.startsWith('127.') } return false