feat: improve enricher rate-limit handling for higher throughput [CM-1220] (#4189)

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

Author: mbani01

services/apps/packages_worker/src/bin/github-repos-enricher.ts

@@ -24,14 +24,23 @@ const main = async () => {
   const enricherConfig = getEnricherConfig()
   const appConfig = getGithubAppConfig()
 
-  const installationIds = await resolveInstallations(appConfig)
+  const discoveredIds = await resolveInstallations(appConfig)
 
-  if (installationIds.length === 0) {
+  if (discoveredIds.length === 0) {
     log.error('No GitHub App installations found — cannot build token pool')
     process.exit(1)
   }
 
-  await fetchRateLimitDiagnostics(appConfig.appId, appConfig.privateKeyPem, installationIds)
+  const installationIds = await fetchRateLimitDiagnostics(
+    appConfig.appId,
+    appConfig.privateKeyPem,
+    discoveredIds,
+  )
+
+  if (installationIds.length === 0) {
+    log.error('All installations failed the rate limit probe — cannot build token pool')
+    process.exit(1)
+  }
 
   log.info(
     {

services/apps/packages_worker/src/config.ts

@@ -41,7 +41,7 @@ export function getEnricherConfig() {
   return {
     updateIntervalHours: requireEnvInt('ENRICHER_REPO_UPDATE_INTERVAL_HOURS'),
     idleSleepSec: requireEnvInt('ENRICHER_IDLE_SLEEP_SEC'),
-    concurrency: parseInt(process.env.ENRICHER_CONCURRENCY ?? '80', 10),
+    concurrency: parseInt(process.env.ENRICHER_CONCURRENCY ?? '150', 10),
     fetchTimeoutMs: parseInt(process.env.ENRICHER_FETCH_TIMEOUT_MS ?? '10000', 10),
   }
 }

services/apps/packages_worker/src/enricher/fetchActivitySnapshot.ts

@@ -100,6 +100,8 @@ function buildDateWindows(): {
 
 interface RateLimit {
   cost: number
+  remaining: number
+  resetAt: string
 }
 
 interface IssueCount {
@@ -164,7 +166,7 @@ const SUMMARY_QUERY = `
     $searchIssuesOpened: String!, $searchIssuesClosed: String!,
     $searchIssuesOpened6m: String!, $searchIssuesOpenNow: String!
   ) {
-    rateLimit { cost }
+    rateLimit { cost remaining resetAt }
     repository(owner: $owner, name: $name) {
       defaultBranchRef {
         target {
@@ -188,7 +190,7 @@ const SUMMARY_QUERY = `
 
 const PR_PAGE_QUERY = `
   query($owner: String!, $name: String!, $cursor: String) {
-    rateLimit { cost }
+    rateLimit { cost remaining resetAt }
     repository(owner: $owner, name: $name) {
       pullRequests(first: ${PAGE_SIZE}, orderBy: { field: CREATED_AT, direction: DESC }, after: $cursor) {
         pageInfo { hasNextPage endCursor }
@@ -210,10 +212,16 @@ async function pagePrs(
   token: string,
   timeoutMs: number,
   windowStart: Date,
-): Promise<{ nodes: PrNode[]; rateLimitCost: number; httpRequestCount: number }> {
+): Promise<{
+  nodes: PrNode[]
+  rateLimitCost: number
+  httpRequestCount: number
+  rateLimit: RateLimit | null
+}> {
   const nodes: PrNode[] = []
   let rateLimitCost = 0
   let httpRequestCount = 0
+  let rateLimit: RateLimit | null = null
   let cursor: string | undefined
 
   for (;;) {
@@ -223,6 +231,7 @@ async function pagePrs(
     const data = await graphqlRequest<PrPageQueryResult>(PR_PAGE_QUERY, variables, token, timeoutMs)
     rateLimitCost += data.rateLimit.cost
     httpRequestCount++
+    rateLimit = data.rateLimit
 
     const connection = data.repository.pullRequests
     let reachedWindowBoundary = false
@@ -239,12 +248,12 @@ async function pagePrs(
     cursor = connection.pageInfo.endCursor
   }
 
-  return { nodes, rateLimitCost, httpRequestCount }
+  return { nodes, rateLimitCost, httpRequestCount, rateLimit }
 }
 
 const ISSUE_PAGE_QUERY = `
   query($owner: String!, $name: String!, $cursor: String) {
-    rateLimit { cost }
+    rateLimit { cost remaining resetAt }
     repository(owner: $owner, name: $name) {
       issues(first: ${PAGE_SIZE}, orderBy: { field: CREATED_AT, direction: DESC }, after: $cursor) {
         pageInfo { hasNextPage endCursor }
@@ -265,10 +274,16 @@ async function pageIssues(
   token: string,
   timeoutMs: number,
   windowStart: Date,
-): Promise<{ nodes: IssueNode[]; rateLimitCost: number; httpRequestCount: number }> {
+): Promise<{
+  nodes: IssueNode[]
+  rateLimitCost: number
+  httpRequestCount: number
+  rateLimit: RateLimit | null
+}> {
   const nodes: IssueNode[] = []
   let rateLimitCost = 0
   let httpRequestCount = 0
+  let rateLimit: RateLimit | null = null
   let cursor: string | undefined
 
   for (;;) {
@@ -283,6 +298,7 @@ async function pageIssues(
     )
     rateLimitCost += data.rateLimit.cost
     httpRequestCount++
+    rateLimit = data.rateLimit
 
     const connection = data.repository.issues
     let reachedWindowBoundary = false
@@ -299,7 +315,7 @@ async function pageIssues(
     cursor = connection.pageInfo.endCursor
   }
 
-  return { nodes, rateLimitCost, httpRequestCount }
+  return { nodes, rateLimitCost, httpRequestCount, rateLimit }
 }
 
 export async function fetchActivitySnapshot(
@@ -363,6 +379,11 @@ export async function fetchActivitySnapshot(
   const prMedians = computePrMedians(prResult.nodes)
   const issueMedians = computeIssueMedians(issueResult.nodes)
 
+  // Lowest remaining across all requests — remaining only decreases within a reset window
+  const lastRateLimit = [summaryData.rateLimit, prResult.rateLimit, issueResult.rateLimit]
+    .filter((rl): rl is RateLimit => rl !== null)
+    .reduce((min, rl) => (rl.remaining < min.remaining ? rl : min))
+
   return {
     repoId,
     snapshotAt: new Date().toISOString(),
@@ -385,5 +406,7 @@ export async function fetchActivitySnapshot(
     issueMedianTimeToFirstResponseHours: issueMedians.medianTimeToFirstResponseHours,
     httpRequestCount: totalHttpRequests,
     rateLimitCost: totalRateLimitCost,
+    rateLimitRemaining: lastRateLimit.remaining,
+    rateLimitResetAt: lastRateLimit.resetAt,
   }
 }

services/apps/packages_worker/src/enricher/fetchLightRepo.ts

@@ -52,6 +52,20 @@ async function fetchSecurityFileEnabled(
       })
       if (response.status === 200) return true
       if (response.status === 404) return false
+      if (response.status === 403) {
+        const body = await response.text()
+        if (body.toLowerCase().includes('rate limit')) {
+          // REST secondary limits send retry-after; primary limits send x-ratelimit-reset
+          const retryAfterSec = parseInt(response.headers.get('retry-after') ?? '0', 10)
+          const resetSec = parseInt(response.headers.get('x-ratelimit-reset') ?? '0', 10)
+          const resetMs = retryAfterSec
+            ? Date.now() + retryAfterSec * 1000
+            : resetSec
+              ? resetSec * 1000 + 5_000
+              : Date.now() + 65_000
+          throw new FetchError('RATE_LIMIT', `Contents API rate limited on ${path}`, resetMs)
+        }
+      }
       throw new Error(`Unexpected status ${response.status} for ${path}`)
     } finally {
       clearTimeout(timeoutId)
@@ -65,6 +79,8 @@ async function fetchSecurityFileEnabled(
     ])
     return root || dotGithub
   } catch (err) {
+    // Rate limits propagate so the caller can park the installation and requeue the repo
+    if (err instanceof FetchError && err.kind === 'RATE_LIMIT') throw err
     log.warn(
       {
         url,

services/apps/packages_worker/src/enricher/githubAppAuth.ts

@@ -2,6 +2,8 @@ import jwt from 'jsonwebtoken'
 
 import { getServiceChildLogger } from '@crowd/logging'
 
+import { FetchError } from './types'
+
 const log = getServiceChildLogger('github-app-auth')
 
 const GITHUB_API = 'https://api.github.com'
@@ -80,7 +82,22 @@ export async function getInstallationToken(
 
   if (!resp.ok) {
     const body = await resp.text()
-    throw new Error(
+    if (resp.status === 429 || (resp.status === 403 && body.toLowerCase().includes('rate limit'))) {
+      const retryAfterSec = parseInt(resp.headers.get('retry-after') ?? '0', 10)
+      const resetSec = parseInt(resp.headers.get('x-ratelimit-reset') ?? '0', 10)
+      const resetMs = retryAfterSec
+        ? Date.now() + retryAfterSec * 1000
+        : resetSec
+          ? resetSec * 1000 + 5_000
+          : Date.now() + 65_000
+      throw new FetchError(
+        'RATE_LIMIT',
+        `Rate limited minting token for installation ${installationId}`,
+        resetMs,
+      )
+    }
+    throw new FetchError(
+      'AUTH',
       `Failed to mint token for installation ${installationId} (${resp.status}): ${body}`,
     )
   }
@@ -100,23 +117,35 @@ interface RateLimitEntry {
 }
 
 /**
- * Fetches GraphQL rate limit info for every installation in parallel (one-time startup diagnostic).
- * GET /rate_limit does not consume quota.
+ * Fetches GraphQL rate limit info for every installation in parallel (one-time startup check).
+ * GET /rate_limit does not consume quota. Excludes installations that fail permanently
+ * (IP allowlists, suspended apps) — those would fail every enrichment request too.
+ * Transient probe failures stay in the pool; runtime parking handles them if truly broken.
  */
 export async function fetchRateLimitDiagnostics(
   appId: string,
   privateKeyPem: string,
   installationIds: number[],
-): Promise<void> {
+): Promise<number[]> {
   const results = await Promise.all(
     installationIds.map(
-      async (id): Promise<RateLimitEntry | { installationId: number; error: string }> => {
+      async (
+        id,
+      ): Promise<
+        RateLimitEntry | { installationId: number; error: string; permanent: boolean }
+      > => {
         try {
           const token = await getInstallationToken(appId, privateKeyPem, id)
           const resp = await fetch(`${GITHUB_API}/rate_limit`, {
             headers: { Authorization: `bearer ${token}`, Accept: 'application/vnd.github+json' },
           })
-          if (!resp.ok) return { installationId: id, error: `HTTP ${resp.status}` }
+          if (!resp.ok) {
+            return {
+              installationId: id,
+              error: `HTTP ${resp.status}`,
+              permanent: [401, 403, 404].includes(resp.status),
+            }
+          }
           // eslint-disable-next-line @typescript-eslint/no-explicit-any
           const data = (await resp.json()) as any
           const g = data.resources?.graphql
@@ -128,34 +157,53 @@ export async function fetchRateLimitDiagnostics(
             resetAt: new Date(g.reset * 1000).toISOString(),
           }
         } catch (err) {
-          return { installationId: id, error: (err as Error).message }
+          const permanent = err instanceof FetchError && err.kind === 'AUTH'
+          return { installationId: id, error: (err as Error).message, permanent }
         }
       },
     ),
   )
 
   let totalLimit = 0
   let totalRemaining = 0
-  let failures = 0
+  const healthyIds: number[] = []
+  const excluded: Array<{ installationId: number; error: string }> = []
+  const transientFailures: Array<{ installationId: number; error: string }> = []
   for (const r of results) {
     if ('error' in r) {
-      failures++
+      if (r.permanent) {
+        excluded.push({ installationId: r.installationId, error: r.error })
+      } else {
+        transientFailures.push({ installationId: r.installationId, error: r.error })
+        healthyIds.push(r.installationId)
+      }
     } else {
       totalLimit += r.limit
       totalRemaining += r.remaining
+      healthyIds.push(r.installationId)
     }
   }
 
   log.info(
     {
       installations: installationIds.length,
-      failures,
+      excluded: excluded.length,
+      transientFailures: transientFailures.length,
       totalLimit,
       totalRemaining,
       totalUsed: totalLimit - totalRemaining,
     },
     'GitHub App rate limit capacity',
   )
+
+  if (transientFailures.length > 0) {
+    log.warn({ transientFailures }, 'Probe failed transiently — keeping installations in the pool')
+  }
+  if (excluded.length > 0) {
+    log.warn({ excluded }, 'Excluding installations that permanently failed the probe')
+  }
+
+  return healthyIds
 }
 
 export interface GithubAppConfig {