chore: raise security contacts worker concurrency; hardcode tuning constants

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

Author: mbani01

backend/.env.dist.composed

@@ -37,8 +37,4 @@ CROWD_PACKAGES_DB_PASSWORD=example
 CROWD_PACKAGES_DB_DATABASE=packages-db
 
 # security-contacts-worker
-SECURITY_CONTACTS_UPDATE_INTERVAL_HOURS=24
-SECURITY_CONTACTS_USER_AGENT="lfx-security-contacts-worker"
-SECURITY_CONTACTS_CONCURRENCY=20
-SECURITY_CONTACTS_FETCH_TIMEOUT_MS=10000
-SECURITY_CONTACTS_BATCH_SIZE=500
+SECURITY_CONTACTS_USER_AGENT="lfx-security-contacts-worker (security@linuxfoundation.org)"

backend/.env.dist.local

@@ -186,11 +186,7 @@ ENRICHER_REPO_UPDATE_INTERVAL_HOURS=24
 ENRICHER_IDLE_SLEEP_SEC=60
 
 # security-contacts-worker
-SECURITY_CONTACTS_UPDATE_INTERVAL_HOURS=24
 SECURITY_CONTACTS_USER_AGENT="lfx-security-contacts-worker (security@linuxfoundation.org)"
-SECURITY_CONTACTS_CONCURRENCY=20
-SECURITY_CONTACTS_FETCH_TIMEOUT_MS=10000
-SECURITY_CONTACTS_BATCH_SIZE=500
 
 OSSPCKGS_GCP_PROJECT=local-dev
 OSSPCKGS_GCS_BUCKET=local-dev

scripts/services/security-contacts-worker.yaml

@@ -7,7 +7,6 @@ x-env-args: &env-args
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
   CROWD_TEMPORAL_TASKQUEUE: packages-worker
-  SECURITY_CONTACTS_UPDATE_INTERVAL_HOURS: '24'
 
 services:
   security-contacts-worker:

services/apps/packages_worker/src/config.ts

@@ -48,14 +48,8 @@ export function getEnricherConfig() {
 
 export function getSecurityContactsConfig() {
   return {
-    // A repo is re-evaluated once its contacts are older than this many hours.
-    updateIntervalHours: requireEnvInt('SECURITY_CONTACTS_UPDATE_INTERVAL_HOURS'),
     // Sent on all registry calls; crates.io rejects requests without an identifying UA.
     userAgent: requireEnv('SECURITY_CONTACTS_USER_AGENT'),
-    // Lower than the enricher: each repo fans out to several HTTP calls across extractors
-    concurrency: parseInt(process.env.SECURITY_CONTACTS_CONCURRENCY ?? '20', 10),
-    fetchTimeoutMs: parseInt(process.env.SECURITY_CONTACTS_FETCH_TIMEOUT_MS ?? '10000', 10),
-    batchSize: parseInt(process.env.SECURITY_CONTACTS_BATCH_SIZE ?? '500', 10),
   }
 }
 

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

@@ -30,6 +30,20 @@ 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; safe against GitHub REST secondary limits given the token pool.
+//  - 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).
+const CONCURRENCY = 100
+const FETCH_TIMEOUT_MS = 15000
+const BATCH_SIZE = 500
+
 const EXTRACTORS: Extractor[] = [
   extractSecurityInsights, // A1
   extractPvr, // A2
@@ -46,7 +60,7 @@ interface SweepRow {
   packages: RepoPackage[] | null
 }
 
-async function fetchBatch(qx: QueryExecutor, config: Config): Promise<SweepRow[]> {
+async function fetchBatch(qx: QueryExecutor): Promise<SweepRow[]> {
   return qx.select(
     `
     SELECT r.id::text AS id,
@@ -58,14 +72,28 @@ async function fetchBatch(qx: QueryExecutor, config: Config): Promise<SweepRow[]
     JOIN packages p ON p.id = pr.package_id AND p.is_critical
     WHERE r.host = 'github'
       AND (
+        -- never evaluated → always eligible
         r.contacts_last_refreshed IS NULL
-        OR r.contacts_last_refreshed < NOW() - INTERVAL '$(updateIntervalHours) hours'
+        -- 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)
+          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)
+          AND r.contacts_last_refreshed < NOW() - INTERVAL '$(weeklyIntervalHours) hours'
+        )
       )
     GROUP BY r.id
     ORDER BY r.id
     LIMIT $(batchSize)
     `,
-    { batchSize: config.batchSize, updateIntervalHours: config.updateIntervalHours },
+    {
+      batchSize: BATCH_SIZE,
+      dailyIntervalHours: DAILY_INTERVAL_HOURS,
+      weeklyIntervalHours: WEEKLY_INTERVAL_HOURS,
+    },
   )
 }
 
@@ -130,17 +158,17 @@ async function processRepo(
 }
 
 export async function processBatch(qx: QueryExecutor, config: Config): Promise<BatchResult> {
-  const batch = await fetchBatch(qx, config)
+  const batch = await fetchBatch(qx)
   if (batch.length === 0) return { processed: 0 }
 
   const deps: ExtractorDeps = {
-    fetchTimeoutMs: config.fetchTimeoutMs,
+    fetchTimeoutMs: FETCH_TIMEOUT_MS,
     userAgent: config.userAgent,
     getToken: getSecurityContactsToken,
   }
 
   const targets = batch.map(toTarget)
-  await runWithConcurrency(targets, config.concurrency, (target) => processRepo(target, deps, qx))
+  await runWithConcurrency(targets, CONCURRENCY, (target) => processRepo(target, deps, qx))
 
   log.info({ processed: targets.length }, 'Security contacts batch complete')
   return { processed: targets.length }