fix: npm 429 retries

services/apps/packages_worker/src/blast-radius/clients/npmAbbreviated.ts

@@ -1,5 +1,5 @@
 import { combineSignals } from '../../npm/signals'
-import type { FetchError } from '../../npm/types'
+import { type FetchError, FetchErrorKind } from '../../npm/types'
 
 const REGISTRY = 'https://registry.npmjs.org'
 const USER_AGENT = 'lfx-packages-worker/0.1 (+https://lfx.linuxfoundation.org)'
@@ -43,21 +43,23 @@ export async function fetchAbbreviatedPackument(
       signal: combinedSignal,
     })
   } catch (err) {
-    return { kind: 'TRANSIENT', message: String(err) }
+    return { kind: FetchErrorKind.TRANSIENT, message: String(err) }
   } finally {
     clearTimeout(timer)
   }
 
   if (res.status === 404)
-    return { kind: 'NOT_FOUND', message: `${name} not found`, statusCode: 404 }
-  if (res.status === 429) return { kind: 'RATE_LIMIT', message: 'rate limited', statusCode: 429 }
-  if (!res.ok) return { kind: 'TRANSIENT', message: `HTTP ${res.status}`, statusCode: res.status }
+    return { kind: FetchErrorKind.NOT_FOUND, message: `${name} not found`, statusCode: 404 }
+  if (res.status === 429)
+    return { kind: FetchErrorKind.RATE_LIMIT, message: 'rate limited', statusCode: 429 }
+  if (!res.ok)
+    return { kind: FetchErrorKind.TRANSIENT, message: `HTTP ${res.status}`, statusCode: res.status }
 
   let json: unknown
   try {
     json = await res.json()
   } catch {
-    return { kind: 'MALFORMED', message: 'invalid JSON' }
+    return { kind: FetchErrorKind.MALFORMED, message: 'invalid JSON' }
   }
 
   if (
@@ -66,7 +68,7 @@ export async function fetchAbbreviatedPackument(
     !('versions' in json) ||
     !('dist-tags' in json)
   ) {
-    return { kind: 'MALFORMED', message: 'unexpected shape' }
+    return { kind: FetchErrorKind.MALFORMED, message: 'unexpected shape' }
   }
 
   return json as AbbreviatedPackument

services/apps/packages_worker/src/blast-radius/dependentsScan.ts

@@ -4,7 +4,7 @@ import * as path from 'path'
 
 import { fetchBulkPointRange, fetchPointRange } from '../npm/fetchDownloads'
 import { fetchPackument } from '../npm/fetchPackument'
-import { FetchError, isFetchError } from '../npm/types'
+import { FetchError, FetchErrorKind, isFetchError } from '../npm/types'
 
 import { fetchAbbreviatedPackument } from './clients/npmAbbreviated'
 import { downloadAndExtractTarball } from './clients/npmTarball'
@@ -120,7 +120,7 @@ async function withRateLimitRetry<T>(
 ): Promise<T | FetchError> {
   let result = await fetchFn()
   for (const delayMs of RATE_LIMIT_RETRY_DELAYS_MS) {
-    if (!isFetchError(result) || result.kind !== 'RATE_LIMIT' || signal?.aborted) break
+    if (!isFetchError(result) || result.kind !== FetchErrorKind.RATE_LIMIT || signal?.aborted) break
     await new Promise((resolve) => setTimeout(resolve, delayMs))
     if (signal?.aborted) break
     result = await fetchFn()

services/apps/packages_worker/src/npm/activities.ts

@@ -1,4 +1,4 @@
-import { Context } from '@temporalio/activity'
+import { ApplicationFailure, Context } from '@temporalio/activity'
 import { type Dispatcher, ProxyAgent } from 'undici'
 
 import {
@@ -11,6 +11,7 @@ import {
   getNpmPurlsDueForLast30dHistory,
   getNpmPurlsDueForLatest30d,
   getNpmPurlsForChangedNames,
+  getNpmPurlsScannedSince,
   getUnscannedNpmPurls,
   insertDailyDownloads,
   logAuditFieldChanges,
@@ -38,7 +39,7 @@ import {
 import { fetchPackument } from './fetchPackument'
 import { Last30dWindow, computeMissingLast30dWindows } from './last30dGaps'
 import { laneCount, proxyForLane } from './proxies'
-import { isFetchError } from './types'
+import { FetchErrorKind, isFetchError } from './types'
 import { upsertPackage } from './upsertPackage'
 
 const log = getServiceChildLogger('npm')
@@ -97,8 +98,9 @@ export async function commitNpmChangesSeq(lastSeq: string): Promise<void> {
 }
 
 // 4xx errors get a few quick in-lane retries with a small linear backoff (1s, 2s),
-// then the package is given up on and marked scanned. 429/5xx/network errors are NOT
-// handled here — they throw and ride Temporal's exponential activity-retry instead.
+// then the package is given up on and marked scanned. 5xx/network errors throw and ride
+// Temporal's exponential activity-retry; 429s throw with the retry scheduled past the
+// server-stated Retry-After window instead.
 const INGEST_4XX_ATTEMPTS = 3
 const INGEST_4XX_BACKOFF_MS = 1000
 
@@ -118,12 +120,22 @@ async function ingestOne(qx: QueryExecutor, purl: string, dispatcher?: Dispatche
       return
     }
 
-    // 429 / 5xx / network → bubble up so Temporal retries the activity with exponential backoff.
+    // 429 → fail the attempt, but schedule the retry past the server-stated penalty window
+    // (npm blocks for ~300s; the default 30/60/120s ladder always lands back inside it).
+    if (packumentResult.kind === FetchErrorKind.RATE_LIMIT) {
+      const delaySec = Math.min(Math.max(packumentResult.retryAfterSec ?? 300, 30), 3600) + 5
+      throw ApplicationFailure.create({
+        message: `Failed to fetch packument for ${name}: ${packumentResult.message}`,
+        nextRetryDelay: `${delaySec}s`,
+      })
+    }
+
+    // 5xx / network → bubble up so Temporal retries the activity with exponential backoff.
     // MALFORMED is permanent (a 200 body that isn't a packument — retrying won't change it),
     // so it takes the quick-retry-then-skip path below instead of poisoning the lane forever.
     if (
       !isClientError(packumentResult.statusCode, packumentResult.kind) &&
-      packumentResult.kind !== 'MALFORMED'
+      packumentResult.kind !== FetchErrorKind.MALFORMED
     ) {
       throw new Error(`Failed to fetch packument for ${name}: ${packumentResult.message}`)
     }
@@ -168,11 +180,21 @@ export async function ingestNpmPackageBatch(purls: string[], laneIndex: number):
   if (purls.length === 0) return
 
   const qx = await getPackagesDb()
+
+  // A retry replays the same purls array — drop what an earlier attempt already settled
+  // (scanned since first schedule), so the replay doesn't re-spend rate budget on re-downloads.
+  let pending = purls
+  const { attempt, scheduledTimestampMs } = Context.current().info
+  if (attempt > 1) {
+    const done = new Set(await getNpmPurlsScannedSince(qx, purls, new Date(scheduledTimestampMs)))
+    if (done.size > 0) pending = purls.filter((p) => !done.has(p))
+  }
+
   const proxy = proxyForLane(laneIndex)
   const dispatcher = proxy ? new ProxyAgent(proxyUrl(proxy)) : undefined
 
   try {
-    for (const purl of purls) {
+    for (const purl of pending) {
       await sleep(ingestSleepMs())
       await ingestOne(qx, purl, dispatcher)
     }
@@ -181,7 +203,12 @@ export async function ingestNpmPackageBatch(purls: string[], laneIndex: number):
   }
 
   log.info(
-    { laneIndex, count: purls.length, exit: proxy?.host ?? 'direct' },
+    {
+      laneIndex,
+      count: pending.length,
+      skipped: purls.length - pending.length,
+      exit: proxy?.host ?? 'direct',
+    },
     'Ingested npm package batch',
   )
 }

services/apps/packages_worker/src/npm/fetchChanges.ts

@@ -1,4 +1,4 @@
-import type { FetchError } from './types'
+import { type FetchError, FetchErrorKind } from './types'
 
 const USER_AGENT = 'lfx-packages-worker/0.1 (+https://lfx.linuxfoundation.org)'
 
@@ -21,21 +21,23 @@ export async function fetchChangesSince(since: string): Promise<ChangesResult |
       signal: abort.signal,
     })
   } catch (err) {
-    return { kind: 'TRANSIENT', message: String(err) }
+    return { kind: FetchErrorKind.TRANSIENT, message: String(err) }
   } finally {
     clearTimeout(timer)
   }
 
-  if (!res.ok) return { kind: 'TRANSIENT', message: `HTTP ${res.status}`, statusCode: res.status }
+  if (!res.ok)
+    return { kind: FetchErrorKind.TRANSIENT, message: `HTTP ${res.status}`, statusCode: res.status }
 
   let body: { results?: unknown[]; last_seq?: unknown }
   try {
     body = (await res.json()) as { results?: unknown[]; last_seq?: unknown }
   } catch {
-    return { kind: 'MALFORMED', message: 'invalid JSON' }
+    return { kind: FetchErrorKind.MALFORMED, message: 'invalid JSON' }
   }
 
-  if (!Array.isArray(body.results)) return { kind: 'MALFORMED', message: 'missing results' }
+  if (!Array.isArray(body.results))
+    return { kind: FetchErrorKind.MALFORMED, message: 'missing results' }
 
   const names = new Set<string>()
   for (const row of body.results as Array<{ id?: unknown }>) {
@@ -58,20 +60,21 @@ export async function fetchCurrentSeq(): Promise<string | FetchError> {
       signal: abort.signal,
     })
   } catch (err) {
-    return { kind: 'TRANSIENT', message: String(err) }
+    return { kind: FetchErrorKind.TRANSIENT, message: String(err) }
   } finally {
     clearTimeout(timer)
   }
 
-  if (!res.ok) return { kind: 'TRANSIENT', message: `HTTP ${res.status}` }
+  if (!res.ok) return { kind: FetchErrorKind.TRANSIENT, message: `HTTP ${res.status}` }
 
   let body: { update_seq?: unknown }
   try {
     body = (await res.json()) as { update_seq?: unknown }
   } catch {
-    return { kind: 'MALFORMED', message: 'invalid JSON' }
+    return { kind: FetchErrorKind.MALFORMED, message: 'invalid JSON' }
   }
 
-  if (body.update_seq === undefined) return { kind: 'MALFORMED', message: 'missing update_seq' }
+  if (body.update_seq === undefined)
+    return { kind: FetchErrorKind.MALFORMED, message: 'missing update_seq' }
   return String(body.update_seq)
 }

services/apps/packages_worker/src/npm/fetchDownloads.ts

@@ -1,7 +1,7 @@
 import type { Dispatcher } from 'undici'
 
 import { combineSignals } from './signals'
-import type { FetchError } from './types'
+import { type FetchError, FetchErrorKind } from './types'
 
 const USER_AGENT = 'lfx-packages-worker/0.1 (+https://lfx.linuxfoundation.org)'
 
@@ -52,34 +52,35 @@ export async function fetchBulkPointRange(
   try {
     res = await fetch(url, downloadInit(dispatcher, combineSignals(abort.signal, signal)))
   } catch (err) {
-    return { kind: 'TRANSIENT', message: String(err) }
+    return { kind: FetchErrorKind.TRANSIENT, message: String(err) }
   } finally {
     clearTimeout(timer)
   }
 
   if (res.status === 404)
-    return { kind: 'NOT_FOUND', message: `bulk request not found`, statusCode: 404 }
+    return { kind: FetchErrorKind.NOT_FOUND, message: `bulk request not found`, statusCode: 404 }
   if (res.status === 429) {
     const headers: Record<string, string> = {}
     for (const [k, v] of res.headers.entries()) headers[k] = v
     const body = await res.text().catch(() => '')
     return {
-      kind: 'RATE_LIMIT',
+      kind: FetchErrorKind.RATE_LIMIT,
       message: `rate limited by npm downloads API — headers: ${JSON.stringify(headers)} body: ${body}`,
       statusCode: 429,
     }
   }
-  if (!res.ok) return { kind: 'TRANSIENT', message: `HTTP ${res.status}`, statusCode: res.status }
+  if (!res.ok)
+    return { kind: FetchErrorKind.TRANSIENT, message: `HTTP ${res.status}`, statusCode: res.status }
 
   let json: unknown
   try {
     json = await res.json()
   } catch {
-    return { kind: 'MALFORMED', message: 'invalid JSON' }
+    return { kind: FetchErrorKind.MALFORMED, message: 'invalid JSON' }
   }
 
   if (typeof json !== 'object' || json === null)
-    return { kind: 'MALFORMED', message: 'expected object response' }
+    return { kind: FetchErrorKind.MALFORMED, message: 'expected object response' }
 
   const raw = json as Record<string, { downloads?: unknown; start?: unknown; end?: unknown } | null>
   const counts = new Map<string, number>()
@@ -111,35 +112,36 @@ export async function fetchPointRange(
   try {
     res = await fetch(url, downloadInit(dispatcher, combineSignals(abort.signal, signal)))
   } catch (err) {
-    return { kind: 'TRANSIENT', message: String(err) }
+    return { kind: FetchErrorKind.TRANSIENT, message: String(err) }
   } finally {
     clearTimeout(timer)
   }
 
   if (res.status === 404)
-    return { kind: 'NOT_FOUND', message: `${name} not found`, statusCode: 404 }
+    return { kind: FetchErrorKind.NOT_FOUND, message: `${name} not found`, statusCode: 404 }
   if (res.status === 429) {
     const headers: Record<string, string> = {}
     for (const [k, v] of res.headers.entries()) headers[k] = v
     const body = await res.text().catch(() => '')
     return {
-      kind: 'RATE_LIMIT',
+      kind: FetchErrorKind.RATE_LIMIT,
       message: `rate limited by npm downloads API — headers: ${JSON.stringify(headers)} body: ${body}`,
       statusCode: 429,
     }
   }
-  if (!res.ok) return { kind: 'TRANSIENT', message: `HTTP ${res.status}`, statusCode: res.status }
+  if (!res.ok)
+    return { kind: FetchErrorKind.TRANSIENT, message: `HTTP ${res.status}`, statusCode: res.status }
 
   let json: unknown
   try {
     json = await res.json()
   } catch {
-    return { kind: 'MALFORMED', message: 'invalid JSON' }
+    return { kind: FetchErrorKind.MALFORMED, message: 'invalid JSON' }
   }
 
   const data = json as { downloads?: unknown; start?: unknown; end?: unknown }
   if (typeof data.downloads !== 'number')
-    return { kind: 'MALFORMED', message: 'missing downloads field' }
+    return { kind: FetchErrorKind.MALFORMED, message: 'missing downloads field' }
 
   return {
     count: data.downloads,
@@ -168,35 +170,36 @@ export async function fetchDailyRange(
   try {
     res = await fetch(url, downloadInit(dispatcher, abort.signal))
   } catch (err) {
-    return { kind: 'TRANSIENT', message: String(err) }
+    return { kind: FetchErrorKind.TRANSIENT, message: String(err) }
   } finally {
     clearTimeout(timer)
   }
 
   if (res.status === 404)
-    return { kind: 'NOT_FOUND', message: `${name} not found`, statusCode: 404 }
+    return { kind: FetchErrorKind.NOT_FOUND, message: `${name} not found`, statusCode: 404 }
   if (res.status === 429) {
     const headers: Record<string, string> = {}
     for (const [k, v] of res.headers.entries()) headers[k] = v
     const body = await res.text().catch(() => '')
     return {
-      kind: 'RATE_LIMIT',
+      kind: FetchErrorKind.RATE_LIMIT,
       message: `rate limited by npm downloads API — headers: ${JSON.stringify(headers)} body: ${body}`,
       statusCode: 429,
     }
   }
-  if (!res.ok) return { kind: 'TRANSIENT', message: `HTTP ${res.status}`, statusCode: res.status }
+  if (!res.ok)
+    return { kind: FetchErrorKind.TRANSIENT, message: `HTTP ${res.status}`, statusCode: res.status }
 
   let json: unknown
   try {
     json = await res.json()
   } catch {
-    return { kind: 'MALFORMED', message: 'invalid JSON' }
+    return { kind: FetchErrorKind.MALFORMED, message: 'invalid JSON' }
   }
 
   const data = json as { start?: unknown; end?: unknown; downloads?: unknown }
   if (!Array.isArray(data.downloads))
-    return { kind: 'MALFORMED', message: 'missing downloads array' }
+    return { kind: FetchErrorKind.MALFORMED, message: 'missing downloads array' }
 
   return {
     start: typeof data.start === 'string' ? data.start : start,