fix: security contacts ingestion improvements [CM-1297]

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;

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<string> | 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<RepoTree> {
+  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<string>()
+    for (const entry of doc.tree) {
+      if (entry.type === 'blob' && typeof entry.path === 'string') paths.add(entry.path)
+    }
+    return { paths }
+  } catch {
+    return { paths: null }
+  }
+}

services/apps/packages_worker/src/security-contacts/extractors/http.ts

@@ -8,7 +8,8 @@ export function registryHeaders(userAgent: string): Record<string, string> {
 // 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.

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: {} }
 

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
 

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)
   }

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

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'
@@ -81,12 +85,12 @@ async function fetchBatch(qx: QueryExecutor): Promise<SweepRow[]> {
         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'
         )
       )
@@ -114,9 +118,19 @@ function toTarget(row: SweepRow): RepoTarget {
 
 async function processRepo(
   target: RepoTarget,
-  deps: ExtractorDeps,
+  baseDeps: Omit<ExtractorDeps, 'repoTree'>,
   qx: QueryExecutor,
 ): Promise<void> {
+  // 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,24 +172,29 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise<B
   const batch = await fetchBatch(qx)
   if (batch.length === 0) return { processed: 0 }
 
-  const deps: ExtractorDeps = {
+  const deps: Omit<ExtractorDeps, 'repoTree'> = {
     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) {
       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()
     }
   })
 

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

@@ -3,37 +3,57 @@ 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<void> {
   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 * * *'],
       },
       policies: {
         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
     }

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<GithubGetResult>
+  /** Default-branch file paths from one git-tree call; `null` means unresolved — probe as before. */
+  repoTree: { paths: Set<string> | null }
 }
 
 export type Extractor = (target: RepoTarget, deps: ExtractorDeps) => Promise<ExtractorResult>

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

@@ -4,6 +4,9 @@ import type * as activities from './activities'
 
 const acts = proxyActivities<typeof activities>({
   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,

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

@@ -1,53 +1,66 @@
 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<void> {
   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.
+//
+// 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,
   contacts: ScoredContact[],
   policies: Partial<RepoPolicies>,
 ): Promise<void> {
   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 },
+    )
 
-    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),
+          })),
+          `(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`,
+        ),
       )
     }
 
     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)