diff --git a/backend/src/api/public/v1/akrites-external/index.ts b/backend/src/api/public/v1/akrites-external/index.ts index 07bed2dd1b..c99fb41497 100644 --- a/backend/src/api/public/v1/akrites-external/index.ts +++ b/backend/src/api/public/v1/akrites-external/index.ts @@ -12,13 +12,15 @@ import { getAkritesExternalContactDetailBatch } from '../packages/getAkritesExte import { getAkritesExternalPackageDetail } from '../packages/getAkritesExternalPackageDetail' import { getAkritesExternalPackageDetailBatch } from '../packages/getAkritesExternalPackageDetailBatch' import { getBlastRadiusJob } from '../packages/getBlastRadiusJob' +import { getBlastRadiusJobBatch } from '../packages/getBlastRadiusJobBatch' import { submitBlastRadiusJob } from '../packages/submitBlastRadiusJob' +import { submitBlastRadiusJobBatch } from '../packages/submitBlastRadiusJobBatch' const rateLimiter = createRateLimiter({ max: 60, windowMs: 60 * 1000 }) // Blast-radius jobs kick off a Temporal workflow per request, so they get their own, // much stricter limiter — configurable via env so it can be tuned without a redeploy. -// Defaults to 5 requests/hour. +// Defaults to 50 requests/hour. const blastRadiusRateLimitMax = Number(process.env.AKRITES_BLAST_RADIUS_RATE_LIMIT_MAX) const blastRadiusRateLimitWindowMs = Number(process.env.AKRITES_BLAST_RADIUS_RATE_LIMIT_WINDOW_MS) @@ -26,7 +28,7 @@ const blastRadiusRateLimiter = createRateLimiter({ max: Number.isSafeInteger(blastRadiusRateLimitMax) && blastRadiusRateLimitMax > 0 ? blastRadiusRateLimitMax - : 5, + : 50, windowMs: Number.isSafeInteger(blastRadiusRateLimitWindowMs) && blastRadiusRateLimitWindowMs > 0 ? blastRadiusRateLimitWindowMs @@ -75,7 +77,22 @@ export function akritesExternalRouter(): Router { const blastRadiusSubRouter = Router() blastRadiusSubRouter.use(requireScopes([SCOPES.READ_PACKAGES])) blastRadiusSubRouter.post('/jobs', blastRadiusRateLimiter, safeWrap(submitBlastRadiusJob)) + // Bulk submit multiplies Temporal workflow starts per request (up to + // MAX_BLAST_RADIUS_JOBS_PER_BATCH), so it sits behind the same strict + // blastRadiusRateLimiter as the single-job route, not the regular one. + blastRadiusSubRouter.post( + /^\/jobs:batch\/?$/, + blastRadiusRateLimiter, + safeWrap(submitBlastRadiusJobBatch), + ) blastRadiusSubRouter.get('/jobs/:analysisId', rateLimiter, safeWrap(getBlastRadiusJob)) + // Bulk poll is read-only, same cost profile as the other batch endpoints, so + // it uses the regular rateLimiter. + blastRadiusSubRouter.post( + /^\/jobs:batch\/poll\/?$/, + rateLimiter, + safeWrap(getBlastRadiusJobBatch), + ) router.use('/blast-radius', blastRadiusSubRouter) return router diff --git a/backend/src/api/public/v1/akrites-external/openapi.yaml b/backend/src/api/public/v1/akrites-external/openapi.yaml index 08327892fa..a1d1ef1540 100644 --- a/backend/src/api/public/v1/akrites-external/openapi.yaml +++ b/backend/src/api/public/v1/akrites-external/openapi.yaml @@ -51,14 +51,16 @@ tags: - name: Blast Radius description: > Advisory reachability analysis — submit (2a) and poll (2b) are both - implemented. Submitting kicks off a Temporal workflow that runs the - npm reachability pipeline (other ecosystems fail fast with - ECOSYSTEM_NOT_SUPPORTED); poll returns job status and, once done, - results. Rate-limited independently of the other akrites-external - endpoints (default 5 requests/hour, configurable via - AKRITES_BLAST_RADIUS_RATE_LIMIT_MAX / _WINDOW_MS). Same interim scope - note as Advisories applies (read:packages, pending a dedicated - read:advisories scope). + implemented, each with a bulk counterpart. Submitting kicks off a + Temporal workflow that runs the npm reachability pipeline (other + ecosystems fail fast with ECOSYSTEM_NOT_SUPPORTED); poll returns job + status and, once done, results. Bulk submit (jobs:batch) is capped at + 20 jobs per request (10 recommended as the default batch size) — each + entry starts its own workflow — and stays behind the same strict rate + limiter as the single-job route; bulk poll + (jobs:batch/poll) is read-only and capped at 100 like the other batch + endpoints. Same interim scope note as Advisories applies (read:packages, + pending a dedicated read:advisories scope). components: securitySchemes: @@ -409,7 +411,10 @@ components: BlastRadiusJobEntry: type: object required: [analysisId, advisoryId, package, ecosystem, status] - description: Response body of 2a — one job per request, never wrapped in an array. + description: > + Response body of 2a for a single submitted job. Returned directly (not + wrapped in an array) by the single-job submit; the batch submit (2a bulk) + returns an array of these under `results` — see BlastRadiusJobBatchResponse. properties: analysisId: type: string @@ -427,7 +432,74 @@ components: status: type: string enum: [pending, running, done, failed] - description: Always pending — the response is returned before the Temporal workflow runs. + description: > + Pending in the single-job submit response, always returned before the + Temporal workflow runs. In the batch submit response, a job whose + workflow failed to start comes back as failed instead — the rest of + the batch is unaffected. + + BlastRadiusJobBatchRequest: + type: object + required: [jobs] + properties: + jobs: + type: array + minItems: 1 + maxItems: 20 + description: > + Capped much lower than the 100-item read batches — each entry + starts its own Temporal workflow, so the batch multiplies + workflow starts (and reachability-analysis cost) per request. + 10 is the recommended default batch size; 20 is the hard limit. + items: + $ref: '#/components/schemas/BlastRadiusJobRequest' + + BlastRadiusJobBatchResponse: + type: object + required: [results] + description: > + Plain array in request order, one entry per submitted job — unlike the + read batches there is no found/not-found case, every job is submitted. + properties: + results: + type: array + items: + $ref: '#/components/schemas/BlastRadiusJobEntry' + + BlastRadiusJobPollBatchRequest: + type: object + required: [analysisIds] + properties: + analysisIds: + type: array + minItems: 1 + maxItems: 100 + items: + type: string + format: uuid + page: + type: integer + minimum: 1 + default: 1 + pageSize: + type: integer + minimum: 1 + maximum: 100 + default: 20 + + BlastRadiusAnalysisBulkEntry: + type: object + required: [requestedAnalysisId, found, analysis] + properties: + requestedAnalysisId: + type: string + found: + type: boolean + analysis: + type: object + nullable: true + allOf: + - $ref: '#/components/schemas/BlastRadiusAnalysis' BlastRadiusResultConfidence: type: string @@ -1045,6 +1117,122 @@ paths: schema: $ref: '#/components/schemas/Error' + /akrites-external/blast-radius/jobs:batch: + post: + operationId: submitBlastRadiusJobBatch + summary: 2a bulk — Submit multiple blast-radius analysis jobs + description: > + One job per array entry, same semantics as the single-job submit — + omit package for an advisory-wide analysis, provide it to narrow to a + single package. Each entry starts its own Temporal workflow, so the + batch is capped at 20 jobs (10 recommended as the default batch + size), much lower than the 100-item read batches, and stays behind + the same strict rate limiter as the single-job route. + + + A per-job failure does not fail the whole batch — that entry comes + back with status: 'failed' and the rest still submit. + tags: [Blast Radius] + security: + - M2MBearer: + - read:packages + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/BlastRadiusJobBatchRequest' + responses: + '202': + description: Jobs accepted, in request order. + content: + application/json: + schema: + $ref: '#/components/schemas/BlastRadiusJobBatchResponse' + '400': + description: Validation error (empty array, >20 items, or an invalid job entry). + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '401': + description: Missing or invalid bearer token. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '403': + description: Token missing read:packages scope. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '429': + description: Too many requests — rate-limited independently of the other endpoints. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + + /akrites-external/blast-radius/jobs:batch/poll: + post: + operationId: getBlastRadiusJobBatch + summary: 2b bulk — Poll multiple blast-radius analysis jobs + description: > + Same found/not-found echo shape as the other batch endpoints: an + unknown analysisId comes back { found: false, analysis: null } + instead of 404ing the whole request. Read-only, so it is rate-limited + the same as the other non-blast-radius endpoints, not the strict + submit limiter. + tags: [Blast Radius] + security: + - M2MBearer: + - read:packages + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/BlastRadiusJobPollBatchRequest' + responses: + '200': + description: One page of results, in request order. + content: + application/json: + schema: + type: object + required: [page, pageSize, total, results] + properties: + page: + type: integer + pageSize: + type: integer + total: + type: integer + description: Total number of requested analysisIds, across all pages. + results: + type: array + items: + $ref: '#/components/schemas/BlastRadiusAnalysisBulkEntry' + '400': + description: Validation error (empty array, >100 items, or a non-uuid analysisId). + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '401': + description: Missing or invalid bearer token. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '403': + description: Token missing read:packages scope. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + /akrites-external/blast-radius/jobs/{analysisId}: get: operationId: getBlastRadiusJob diff --git a/backend/src/api/public/v1/packages/blastRadiusBatch.test.ts b/backend/src/api/public/v1/packages/blastRadiusBatch.test.ts new file mode 100644 index 0000000000..b7201336fa --- /dev/null +++ b/backend/src/api/public/v1/packages/blastRadiusBatch.test.ts @@ -0,0 +1,93 @@ +import { describe, expect, it } from 'vitest' + +import { + MAX_BLAST_RADIUS_JOBS_PER_BATCH, + MAX_BLAST_RADIUS_POLL_IDS_PER_BATCH, + blastRadiusJobBatchRequestSchema, + blastRadiusJobPollBatchRequestSchema, + paginateAnalysisIds, +} from './blastRadiusBatch' + +describe('blastRadiusJobBatchRequestSchema', () => { + it('accepts a batch of valid job requests', () => { + const result = blastRadiusJobBatchRequestSchema.safeParse({ + jobs: [ + { advisoryId: 'GHSA-jf85-cpcp-j695', ecosystem: 'npm' }, + { advisoryId: 'CVE-2024-12345', ecosystem: 'npm', package: 'pkg:npm/lodash' }, + ], + }) + expect(result.success).toBe(true) + }) + + it('rejects an empty jobs array', () => { + const result = blastRadiusJobBatchRequestSchema.safeParse({ jobs: [] }) + expect(result.success).toBe(false) + }) + + it('rejects more than MAX_BLAST_RADIUS_JOBS_PER_BATCH jobs', () => { + const jobs = Array.from({ length: MAX_BLAST_RADIUS_JOBS_PER_BATCH + 1 }, () => ({ + advisoryId: 'GHSA-jf85-cpcp-j695', + ecosystem: 'npm', + })) + const result = blastRadiusJobBatchRequestSchema.safeParse({ jobs }) + expect(result.success).toBe(false) + }) + + it('rejects a batch containing one invalid job', () => { + const result = blastRadiusJobBatchRequestSchema.safeParse({ + jobs: [ + { advisoryId: 'GHSA-jf85-cpcp-j695', ecosystem: 'npm' }, + { advisoryId: 'not-an-advisory-id', ecosystem: 'npm' }, + ], + }) + expect(result.success).toBe(false) + }) +}) + +describe('blastRadiusJobPollBatchRequestSchema', () => { + const validId = '3fa85f64-5717-4562-b3fc-2c963f66afa6' + + it('accepts a batch of valid analysisIds and defaults page/pageSize', () => { + const result = blastRadiusJobPollBatchRequestSchema.parse({ analysisIds: [validId] }) + expect(result.page).toBe(1) + expect(result.pageSize).toBe(20) + }) + + it('rejects an empty analysisIds array', () => { + const result = blastRadiusJobPollBatchRequestSchema.safeParse({ analysisIds: [] }) + expect(result.success).toBe(false) + }) + + it('rejects a non-uuid analysisId', () => { + const result = blastRadiusJobPollBatchRequestSchema.safeParse({ analysisIds: ['not-a-uuid'] }) + expect(result.success).toBe(false) + }) + + it('rejects more than MAX_BLAST_RADIUS_POLL_IDS_PER_BATCH analysisIds', () => { + const analysisIds = Array.from( + { length: MAX_BLAST_RADIUS_POLL_IDS_PER_BATCH + 1 }, + () => validId, + ) + const result = blastRadiusJobPollBatchRequestSchema.safeParse({ analysisIds }) + expect(result.success).toBe(false) + }) +}) + +describe('paginateAnalysisIds', () => { + it('slices the requested page out of the full analysisIds array', () => { + const analysisIds = ['a', 'b', 'c', 'd', 'e'] + const result = paginateAnalysisIds({ analysisIds, page: 2, pageSize: 2 }) + expect(result).toEqual({ + page: 2, + pageSize: 2, + total: 5, + pagedAnalysisIds: ['c', 'd'], + }) + }) + + it('returns an empty page past the end of the array', () => { + const result = paginateAnalysisIds({ analysisIds: ['a'], page: 2, pageSize: 20 }) + expect(result.pagedAnalysisIds).toEqual([]) + expect(result.total).toBe(1) + }) +}) diff --git a/backend/src/api/public/v1/packages/blastRadiusBatch.ts b/backend/src/api/public/v1/packages/blastRadiusBatch.ts new file mode 100644 index 0000000000..7989d722d7 --- /dev/null +++ b/backend/src/api/public/v1/packages/blastRadiusBatch.ts @@ -0,0 +1,83 @@ +import { z } from 'zod' + +import { blastRadiusJobRequestSchema } from './blastRadius' +import type { BlastRadiusAnalysis } from './blastRadiusAnalysis' +import { DEFAULT_BATCH_PAGE_SIZE } from './purl' + +// Read-only batches (packages/advisories/contacts) cap at 100 — cheap indexed +// lookups. Batch submit multiplies Temporal workflow starts (and their LLM +// reachability cost) per request, so it gets a much lower cap, independent of +// that constant. 20 is the hard limit agreed with Joana after the cost test +// (2026-07-23) — 10 is the recommended/default batch size for callers, not a +// separate enforced value, since `jobs` is an explicit client-provided array. +export const MAX_BLAST_RADIUS_JOBS_PER_BATCH = 20 + +// Polling is read-only, same cost profile as the other batches, so it reuses +// their 100 cap. +export const MAX_BLAST_RADIUS_POLL_IDS_PER_BATCH = 100 + +export const blastRadiusJobBatchRequestSchema = z.object({ + jobs: z + .array(blastRadiusJobRequestSchema) + .min(1) + .max( + MAX_BLAST_RADIUS_JOBS_PER_BATCH, + `Maximum ${MAX_BLAST_RADIUS_JOBS_PER_BATCH} jobs per request`, + ), +}) + +export type BlastRadiusJobBatchRequest = z.infer + +// Unlike the read batches (purls that may or may not resolve to a package), every +// job in a submit batch is genuinely submitted — there is no "not found" case, so +// the response is a plain array in request order, not a found/not-found wrapper. +const analysisIdSchema = z.uuid() + +export const blastRadiusJobPollBatchRequestSchema = z.object({ + analysisIds: z + .array(analysisIdSchema) + .min(1) + .max( + MAX_BLAST_RADIUS_POLL_IDS_PER_BATCH, + `Maximum ${MAX_BLAST_RADIUS_POLL_IDS_PER_BATCH} analysisIds per request`, + ), + page: z.coerce.number().int().min(1).default(1), + pageSize: z.coerce + .number() + .int() + .min(1) + .max(MAX_BLAST_RADIUS_POLL_IDS_PER_BATCH) + .default(DEFAULT_BATCH_PAGE_SIZE), +}) + +export type BlastRadiusJobPollBatchRequest = z.infer + +export interface BlastRadiusAnalysisBulkEntry { + requestedAnalysisId: string + found: boolean + analysis: BlastRadiusAnalysis | null +} + +export interface PaginatedAnalysisIds { + page: number + pageSize: number + total: number + pagedAnalysisIds: string[] +} + +// Mirrors paginatePurls (purl.ts): slice the requested page out of the full +// analysisIds array. No normalization step — analysisIds are UUIDs, unlike purls. +export function paginateAnalysisIds(body: { + analysisIds: string[] + page: number + pageSize: number +}): PaginatedAnalysisIds { + const { analysisIds, page, pageSize } = body + const start = (page - 1) * pageSize + return { + page, + pageSize, + total: analysisIds.length, + pagedAnalysisIds: analysisIds.slice(start, start + pageSize), + } +} diff --git a/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.test.ts b/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.test.ts new file mode 100644 index 0000000000..f1953cea41 --- /dev/null +++ b/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.test.ts @@ -0,0 +1,157 @@ +import type { Request, Response } from 'express' +import { describe, expect, it, vi } from 'vitest' + +import { getBlastRadiusJobBatch } from './getBlastRadiusJobBatch' + +const { getAnalysisDetailsByIds, getVerdictResultsBatch, getDependentsExcludedByRangeCountBatch } = + vi.hoisted(() => ({ + getAnalysisDetailsByIds: vi.fn(), + getVerdictResultsBatch: vi.fn(), + getDependentsExcludedByRangeCountBatch: vi.fn(), + })) + +vi.mock('@crowd/data-access-layer/src/packages/blastRadius', () => ({ + getAnalysisDetailsByIds, + getVerdictResultsBatch, + getDependentsExcludedByRangeCountBatch, +})) + +vi.mock('@/db/packagesDb', () => ({ + getPackagesQx: vi.fn().mockResolvedValue({}), +})) + +const PENDING_ID = '11111111-1111-4111-8111-111111111111' +const DONE_ID = '22222222-2222-4222-8222-222222222222' +const MISSING_ID = '33333333-3333-4333-8333-333333333333' + +function mockReqRes(body: unknown) { + getAnalysisDetailsByIds.mockClear() + getVerdictResultsBatch.mockClear() + getDependentsExcludedByRangeCountBatch.mockClear() + getVerdictResultsBatch.mockResolvedValue([]) + getDependentsExcludedByRangeCountBatch.mockResolvedValue([]) + + const req = { body } as unknown as Request + + const json = vi.fn() + const status = vi.fn().mockReturnValue({ json }) + const res = { status, json } as unknown as Response + + return { req, res, json } +} + +describe('getBlastRadiusJobBatch', () => { + it('returns found/not-found entries in request order with page metadata', async () => { + getAnalysisDetailsByIds.mockResolvedValue([ + { + id: PENDING_ID, + advisory_osv_id: 'GHSA-jf85-cpcp-j695', + package_name: null, + ecosystem: 'npm', + status: 'pending', + error: null, + candidates_considered: null, + started_at: '2026-07-01T00:00:00.000Z', + completed_at: null, + }, + ]) + + const { req, res, json } = mockReqRes({ + analysisIds: [PENDING_ID, MISSING_ID], + }) + + await getBlastRadiusJobBatch(req, res) + + expect(json).toHaveBeenCalledWith( + expect.objectContaining({ + page: 1, + pageSize: expect.any(Number), + total: 2, + results: [ + expect.objectContaining({ + requestedAnalysisId: PENDING_ID, + found: true, + analysis: expect.objectContaining({ analysisId: PENDING_ID, status: 'pending' }), + }), + { requestedAnalysisId: MISSING_ID, found: false, analysis: null }, + ], + }), + ) + }) + + it('only fetches verdicts/excluded counts for done analyses', async () => { + const { req, res, json } = mockReqRes({ analysisIds: [PENDING_ID, DONE_ID] }) + + getAnalysisDetailsByIds.mockResolvedValue([ + { + id: PENDING_ID, + advisory_osv_id: 'GHSA-jf85-cpcp-j695', + package_name: null, + ecosystem: 'npm', + status: 'pending', + error: null, + candidates_considered: null, + started_at: '2026-07-01T00:00:00.000Z', + completed_at: null, + }, + { + id: DONE_ID, + advisory_osv_id: 'GHSA-652q-gvq3-74qv', + package_name: 'lodash', + ecosystem: 'npm', + status: 'done', + error: null, + candidates_considered: 10, + started_at: '2026-07-01T00:00:00.000Z', + completed_at: '2026-07-01T01:00:00.000Z', + }, + ]) + getVerdictResultsBatch.mockResolvedValue([ + { + analysisId: DONE_ID, + name: 'benchmark.js', + version: '2.1.4', + downloads: 500000, + reachable_verdict: 'affected', + confidence: 0.9, + evidence: null, + reasoning: 'uses merge', + }, + ]) + getDependentsExcludedByRangeCountBatch.mockResolvedValue([{ analysisId: DONE_ID, count: 8 }]) + + await getBlastRadiusJobBatch(req, res) + + expect(getVerdictResultsBatch).toHaveBeenCalledWith(expect.anything(), [DONE_ID]) + expect(getDependentsExcludedByRangeCountBatch).toHaveBeenCalledWith(expect.anything(), [ + DONE_ID, + ]) + + const [{ results }] = json.mock.calls[0] + expect(results[0]).toMatchObject({ requestedAnalysisId: PENDING_ID, found: true }) + expect(results[0].analysis).toMatchObject({ status: 'pending', summary: null, results: null }) + expect(results[1]).toMatchObject({ requestedAnalysisId: DONE_ID, found: true }) + expect(results[1].analysis).toMatchObject({ + status: 'done', + summary: expect.objectContaining({ dependentsExcludedUpfront: 8 }), + }) + }) + + it('rejects a batch with a malformed uuid without querying the database', async () => { + const { req, res } = mockReqRes({ analysisIds: [PENDING_ID, 'not-a-uuid'] }) + + await expect(getBlastRadiusJobBatch(req, res)).rejects.toThrow() + expect(getAnalysisDetailsByIds).not.toHaveBeenCalled() + }) + + it('rejects a batch with more than 100 analysisIds', async () => { + const analysisIds = Array.from( + { length: 101 }, + (_, i) => `44444444-4444-4444-8444-${String(i).padStart(12, '0')}`, + ) + const { req, res } = mockReqRes({ analysisIds }) + + await expect(getBlastRadiusJobBatch(req, res)).rejects.toThrow() + expect(getAnalysisDetailsByIds).not.toHaveBeenCalled() + }) +}) diff --git a/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts b/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts new file mode 100644 index 0000000000..2dd397be19 --- /dev/null +++ b/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts @@ -0,0 +1,68 @@ +import type { Request, Response } from 'express' + +import * as blastRadiusDal from '@crowd/data-access-layer/src/packages/blastRadius' + +import { getPackagesQx } from '@/db/packagesDb' +import { ok } from '@/utils/api' +import { validateOrThrow } from '@/utils/validation' + +import { toBlastRadiusAnalysis } from './blastRadiusAnalysis' +import { + type BlastRadiusAnalysisBulkEntry, + blastRadiusJobPollBatchRequestSchema, + paginateAnalysisIds, +} from './blastRadiusBatch' + +// 2b bulk — poll multiple blast-radius analyses in one request. Same +// found/not-found echo shape as the other batch endpoints (packages, advisories, +// contacts): an unknown analysisId comes back { found: false, analysis: null } +// instead of 404ing the whole request. Read-only, so it stays behind the +// regular rateLimiter, not the strict blastRadiusRateLimiter. +export async function getBlastRadiusJobBatch(req: Request, res: Response): Promise { + const { page, pageSize, total, pagedAnalysisIds } = paginateAnalysisIds( + validateOrThrow(blastRadiusJobPollBatchRequestSchema, req.body), + ) + + const qx = await getPackagesQx() + + const analysisRows = await blastRadiusDal.getAnalysisDetailsByIds(qx, pagedAnalysisIds) + const analysisById = new Map(analysisRows.map((row) => [row.id, row])) + + const doneIds = analysisRows.filter((row) => row.status === 'done').map((row) => row.id) + const [verdictRows, excludedByRangeCounts] = await Promise.all([ + blastRadiusDal.getVerdictResultsBatch(qx, doneIds), + blastRadiusDal.getDependentsExcludedByRangeCountBatch(qx, doneIds), + ]) + + const verdictsByAnalysisId = new Map() + for (const row of verdictRows) { + const bucket = verdictsByAnalysisId.get(row.analysisId) + if (bucket) { + bucket.push(row) + } else { + verdictsByAnalysisId.set(row.analysisId, [row]) + } + } + const excludedByRangeCountByAnalysisId = new Map( + excludedByRangeCounts.map(({ analysisId, count }) => [analysisId, count]), + ) + + const results: BlastRadiusAnalysisBulkEntry[] = pagedAnalysisIds.map((requestedAnalysisId) => { + const analysis = analysisById.get(requestedAnalysisId) + if (!analysis) { + return { requestedAnalysisId, found: false, analysis: null } + } + + return { + requestedAnalysisId, + found: true, + analysis: toBlastRadiusAnalysis( + analysis, + verdictsByAnalysisId.get(requestedAnalysisId) ?? [], + excludedByRangeCountByAnalysisId.get(requestedAnalysisId) ?? 0, + ), + } + }) + + ok(res, { page, pageSize, total, results }) +} diff --git a/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.test.ts b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.test.ts new file mode 100644 index 0000000000..e3f37ae73d --- /dev/null +++ b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.test.ts @@ -0,0 +1,124 @@ +import type { Request, Response } from 'express' +import { describe, expect, it, vi } from 'vitest' + +import { submitBlastRadiusJobBatch } from './submitBlastRadiusJobBatch' + +const { start, createAnalysis, failAnalysis } = vi.hoisted(() => ({ + start: vi.fn().mockResolvedValue(undefined), + createAnalysis: vi.fn().mockResolvedValue(undefined), + failAnalysis: vi.fn().mockResolvedValue(undefined), +})) + +vi.mock('@/db/packagesTemporal', () => ({ + getPackagesTemporalClient: vi.fn().mockResolvedValue({ workflow: { start } }), +})) + +vi.mock('@/db/packagesDb', () => ({ + getPackagesQx: vi.fn().mockResolvedValue({}), +})) + +vi.mock('@crowd/data-access-layer/src/packages/blastRadius', () => ({ + createAnalysis, + failAnalysis, +})) + +function mockReqRes(body: unknown) { + start.mockClear() + createAnalysis.mockClear() + failAnalysis.mockClear() + + const req = { body } as unknown as Request + + const json = vi.fn() + const status = vi.fn().mockReturnValue({ json }) + const res = { status } as unknown as Response + + return { req, res, start, status, json } +} + +describe('submitBlastRadiusJobBatch', () => { + it('starts one workflow per job and responds 202 with results in request order', async () => { + const { req, res, status, json } = mockReqRes({ + jobs: [ + { advisoryId: 'GHSA-jf85-cpcp-j695', ecosystem: 'npm' }, + { advisoryId: 'GHSA-652q-gvq3-74qv', package: 'pkg:npm/lodash', ecosystem: 'npm' }, + ], + }) + + await submitBlastRadiusJobBatch(req, res) + + expect(createAnalysis).toHaveBeenCalledTimes(2) + expect(start).toHaveBeenCalledTimes(2) + expect(status).toHaveBeenCalledWith(202) + + const [{ results }] = json.mock.calls[0] + expect(results).toHaveLength(2) + expect(results[0]).toMatchObject({ + advisoryId: 'GHSA-jf85-cpcp-j695', + package: null, + ecosystem: 'npm', + status: 'pending', + }) + expect(results[1]).toMatchObject({ + advisoryId: 'GHSA-652q-gvq3-74qv', + package: 'pkg:npm/lodash', + ecosystem: 'npm', + status: 'pending', + }) + expect(typeof results[0].analysisId).toBe('string') + expect(typeof results[1].analysisId).toBe('string') + }) + + it('isolates a per-job workflow.start failure to that job only', async () => { + const { req, res, json } = mockReqRes({ + jobs: [ + { advisoryId: 'GHSA-jf85-cpcp-j695', ecosystem: 'npm' }, + { advisoryId: 'GHSA-652q-gvq3-74qv', ecosystem: 'npm' }, + ], + }) + start.mockResolvedValueOnce(undefined).mockRejectedValueOnce(new Error('temporal unreachable')) + + await submitBlastRadiusJobBatch(req, res) + + const [{ results }] = json.mock.calls[0] + expect(results).toHaveLength(2) + expect(results[0]).toMatchObject({ advisoryId: 'GHSA-jf85-cpcp-j695', status: 'pending' }) + expect(results[1]).toMatchObject({ advisoryId: 'GHSA-652q-gvq3-74qv', status: 'failed' }) + + expect(failAnalysis).toHaveBeenCalledTimes(1) + const [, , errorMessage] = failAnalysis.mock.calls[0] + expect(errorMessage).toBe('temporal unreachable') + }) + + it('rejects a batch containing an unsupported ecosystem without submitting any job', async () => { + const { req, res, start } = mockReqRes({ + jobs: [ + { advisoryId: 'GHSA-jf85-cpcp-j695', ecosystem: 'npm' }, + { advisoryId: 'GHSA-652q-gvq3-74qv', ecosystem: 'maven' }, + ], + }) + + await expect(submitBlastRadiusJobBatch(req, res)).rejects.toThrow() + expect(start).not.toHaveBeenCalled() + expect(createAnalysis).not.toHaveBeenCalled() + }) + + it('rejects a batch with more than 20 jobs without submitting any job', async () => { + const jobs = Array.from({ length: 21 }, () => ({ + advisoryId: 'GHSA-jf85-cpcp-j695', + ecosystem: 'npm', + })) + const { req, res, start } = mockReqRes({ jobs }) + + await expect(submitBlastRadiusJobBatch(req, res)).rejects.toThrow() + expect(start).not.toHaveBeenCalled() + expect(createAnalysis).not.toHaveBeenCalled() + }) + + it('rejects an empty jobs array', async () => { + const { req, res, start } = mockReqRes({ jobs: [] }) + + await expect(submitBlastRadiusJobBatch(req, res)).rejects.toThrow() + expect(start).not.toHaveBeenCalled() + }) +}) diff --git a/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts new file mode 100644 index 0000000000..fa0d7b6f89 --- /dev/null +++ b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts @@ -0,0 +1,102 @@ +import type { Request, Response } from 'express' + +import { generateUUIDv4 } from '@crowd/common' +import * as blastRadiusDal from '@crowd/data-access-layer/src/packages/blastRadius' +import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' +import { Client } from '@crowd/temporal' +import { ITriggerBlastRadiusAnalysis, TemporalWorkflowId } from '@crowd/types' + +import { getPackagesQx } from '@/db/packagesDb' +import { getPackagesTemporalClient } from '@/db/packagesTemporal' +import { validateOrThrow } from '@/utils/validation' + +import { + type BlastRadiusJobEntry, + type BlastRadiusJobRequest, + toBlastRadiusJobEntry, +} from './blastRadius' +import { blastRadiusJobBatchRequestSchema } from './blastRadiusBatch' + +// 2a bulk — submit multiple blast-radius analysis jobs in one request, one per +// array entry. Same lifecycle as the single-job submit, just looped: each entry +// gets its own analysisId, its own pending row, and its own Temporal workflow +// start. Unlike the read-only batch endpoints (packages/advisories/contacts), +// this multiplies workflow starts per request, so the batch size is capped much +// lower (see MAX_BLAST_RADIUS_JOBS_PER_BATCH) and the route stays behind the same +// strict blastRadiusRateLimiter as the single-job route. +// +// A per-job failure (e.g. workflow.start throwing) does not fail the whole +// batch — that job's entry comes back status: 'failed' and the rest still +// submit, matching the partial-result shape of the other batch endpoints. +export async function submitBlastRadiusJobBatch(req: Request, res: Response): Promise { + const { jobs } = validateOrThrow(blastRadiusJobBatchRequestSchema, req.body) + + const qx = await getPackagesQx() + const packagesTemporal = await getPackagesTemporalClient() + + const results: BlastRadiusJobEntry[] = await Promise.all( + jobs.map((body) => submitOneJob(qx, packagesTemporal, body)), + ) + + res.status(202).json({ results }) +} + +async function submitOneJob( + qx: QueryExecutor, + packagesTemporal: Client, + body: BlastRadiusJobRequest, +): Promise { + const jobPackage = body.package ?? null + const jobEcosystem = body.ecosystem + const analysisId = generateUUIDv4() + const analysisInput = { + id: analysisId, + advisoryOsvId: body.advisoryId, + packageName: jobPackage, + ecosystem: jobEcosystem, + force: body.force, + } + + try { + // Create the pending row synchronously, before starting the workflow — see the + // same comment on submitBlastRadiusJob for why (avoids a poll-race 404). This is + // inside the try too — unlike the single-job submit, a createAnalysis failure + // must not reject the whole batch's Promise.all, only this job's entry. + await blastRadiusDal.createAnalysis(qx, analysisInput) + + await packagesTemporal.workflow.start('analyzeBlastRadius', { + taskQueue: 'blast-radius-worker', + workflowId: `${TemporalWorkflowId.BLAST_RADIUS_ANALYSIS}/${analysisId}`, + retry: { maximumAttempts: 1 }, + args: [ + { + analysisId, + advisoryId: body.advisoryId, + package: jobPackage, + ecosystem: jobEcosystem, + force: body.force, + } satisfies ITriggerBlastRadiusAnalysis, + ], + }) + + return toBlastRadiusJobEntry({ + analysisId, + advisoryId: body.advisoryId, + package: jobPackage, + ecosystem: jobEcosystem, + }) + } catch (err) { + // Unlike the single-job submit, this does not rethrow — one job's workflow + // failing to start must not take the rest of the batch down with it. + const errorMessage = err instanceof Error ? err.message : String(err) + await blastRadiusDal.failAnalysis(qx, analysisInput, errorMessage) + + return { + analysisId, + advisoryId: body.advisoryId, + package: jobPackage, + ecosystem: jobEcosystem, + status: 'failed', + } + } +} diff --git a/services/apps/packages_worker/src/bin/blast-radius-worker.ts b/services/apps/packages_worker/src/bin/blast-radius-worker.ts index 7e71eeaf35..bad76dbedf 100644 --- a/services/apps/packages_worker/src/bin/blast-radius-worker.ts +++ b/services/apps/packages_worker/src/bin/blast-radius-worker.ts @@ -1,4 +1,25 @@ -import { svc } from '../service' +import { Config } from '@crowd/archetype-standard' +import { Options, ServiceWorker } from '@crowd/archetype-worker' + +// Own ServiceWorker instance rather than the shared one in ../service (used by every +// other packages_worker entry point) — its activities are CPU/IO-heavy (tarball +// downloads+extraction, agent subprocesses), so it needs a concurrency cap that would +// otherwise throttle unrelated, lightweight ingestion workers if set on the shared +// instance. Left uncapped, the SDK default lets dozens run at once in this one process, +// starving the event loop and missing heartbeat deadlines. +const config: Config = { + envvars: [], + producer: { enabled: false }, + temporal: { enabled: true }, + redis: { enabled: false }, +} + +const options: Options = { + postgres: { enabled: false }, // packages-db is managed via getPackagesDb() + maxConcurrentActivityTaskExecutions: 8, +} + +const svc = new ServiceWorker(config, options) // On-demand only — analyzeBlastRadius is triggered per request from the backend's // submitBlastRadiusJob handler (workflow.start), not on a schedule. diff --git a/services/apps/packages_worker/src/blast-radius/dependentsScan.ts b/services/apps/packages_worker/src/blast-radius/dependentsScan.ts index e9def602f6..7f6a5b7317 100644 --- a/services/apps/packages_worker/src/blast-radius/dependentsScan.ts +++ b/services/apps/packages_worker/src/blast-radius/dependentsScan.ts @@ -248,6 +248,10 @@ export async function scanDependents(input: { const unscopedNames = candidateNames.filter((n) => !n.startsWith('@')) const scopedNames = candidateNames.filter((n) => n.startsWith('@')) + // Neither loop below heartbeated before — for a large candidate pool (many + // unscoped batches, or many scoped names run one HTTP call at a time through + // mapWithConcurrency) this whole download-count phase could run past the + // dependents stage's heartbeatTimeout with nothing heartbeating in between. for (let i = 0; i < unscopedNames.length; i += 128) { const batch = unscopedNames.slice(i, i + 128) const result = await fetchBulkPointRange(batch, isoDate(rangeStart), isoDate(rangeEnd)) @@ -256,15 +260,21 @@ export async function scanDependents(input: { downloads.set(name, count) }) } + onProgress?.() } // fetchBulkPointRange doesn't support scoped package names; fetch those individually // so scoped candidates aren't silently ranked at 0 downloads and excluded from topN. + let scopedProcessed = 0 await mapWithConcurrency(scopedNames, SCAN_CONCURRENCY, async (name) => { const result = await fetchPointRange(name, isoDate(rangeStart), isoDate(rangeEnd)) if (!isFetchError(result)) { downloads.set(name, result.count) } + scopedProcessed++ + if (onProgress && scopedProcessed % 200 === 0) { + onProgress() + } }) // Phase 2: walk candidates ranked by downloads (descending), fetching the full packument @@ -279,6 +289,10 @@ export async function scanDependents(input: { for (const name of ranked) { if (analyzed.length >= topN) break + // Phase 1 heartbeats every 200 candidates via mapWithConcurrency; this walk is + // sequential (one packument fetch at a time), so it needs its own heartbeat too. + onProgress?.() + const packument = await fetchPackument(name) if (isFetchError(packument)) continue diff --git a/services/apps/packages_worker/src/blast-radius/stages/reachability.ts b/services/apps/packages_worker/src/blast-radius/stages/reachability.ts index 01e220f2e7..2f9c75ae8f 100644 --- a/services/apps/packages_worker/src/blast-radius/stages/reachability.ts +++ b/services/apps/packages_worker/src/blast-radius/stages/reachability.ts @@ -176,9 +176,13 @@ export async function runReachabilityStage( return } - // Backoff before retry + // Backoff before retry. Heartbeat partway through — the longest backoff + // (attempt 2, 30s) is still well under the stage's 5-minute heartbeatTimeout + // on its own, but 4 dependents processing concurrently can each be mid-backoff + // at once with nothing else heartbeating in between. const delayMs = RETRY_BACKOFF_BASE * attempt await new Promise((resolve) => setTimeout(resolve, delayMs)) + onProgress?.() } } } finally { diff --git a/services/apps/packages_worker/src/blast-radius/workflows.ts b/services/apps/packages_worker/src/blast-radius/workflows.ts index dff8334a21..fdaf2fc87f 100644 --- a/services/apps/packages_worker/src/blast-radius/workflows.ts +++ b/services/apps/packages_worker/src/blast-radius/workflows.ts @@ -38,6 +38,7 @@ const { blastRadiusReachability } = proxyActivities({ const { blastRadiusReport } = proxyActivities({ startToCloseTimeout: '2 minutes', + heartbeatTimeout: '1 minute', retry: { maximumAttempts: 3 }, }) diff --git a/services/libs/data-access-layer/src/packages/blastRadius.ts b/services/libs/data-access-layer/src/packages/blastRadius.ts index f71cf36f92..030657f9ce 100644 --- a/services/libs/data-access-layer/src/packages/blastRadius.ts +++ b/services/libs/data-access-layer/src/packages/blastRadius.ts @@ -177,6 +177,28 @@ export async function getAnalysisDetail( ) } +// Bulk counterpart of getAnalysisDetail for batch polling — one query for the whole +// page instead of one per id. Order is not guaranteed to match analysisIds; callers +// key the result by row.id. +export async function getAnalysisDetailsByIds( + qx: QueryExecutor, + analysisIds: string[], +): Promise { + if (analysisIds.length === 0) { + return [] + } + return qx.select( + ` + SELECT + id, advisory_osv_id, package_name, ecosystem, status, error, + candidates_considered, started_at, completed_at + FROM blast_radius_analyses + WHERE id = ANY($(analysisIds)::uuid[]) + `, + { analysisIds }, + ) +} + export async function setDependentsMeta( qx: QueryExecutor, analysisId: string, @@ -374,6 +396,31 @@ export async function getDependentsExcludedByRangeCount( return Number(row.count) || 0 } +// Bulk counterpart of getDependentsExcludedByRangeCount — one grouped query for the +// whole page instead of one per id. Ids with no excluded rows are simply absent from +// the result; callers should default to 0 for those. +export async function getDependentsExcludedByRangeCountBatch( + qx: QueryExecutor, + analysisIds: string[], +): Promise<{ analysisId: string; count: number }[]> { + if (analysisIds.length === 0) { + return [] + } + const rows = await qx.select( + ` + SELECT analysis_id, COUNT(*) as count + FROM blast_radius_dependents + WHERE analysis_id = ANY($(analysisIds)::uuid[]) AND excluded_by_range = TRUE + GROUP BY analysis_id + `, + { analysisIds }, + ) + return rows.map((row: Record) => ({ + analysisId: row.analysis_id as string, + count: Number(row.count) || 0, + })) +} + // Clears prior dependents for this analysis before a fresh scan re-inserts. Needed // because a retry (stage failed after insertDependents but before completeStageRun) // would otherwise leave stale rows from the earlier attempt's scan around — upserting @@ -491,6 +538,34 @@ export async function getVerdictResults( ) } +// Bulk counterpart of getVerdictResults — one query for the whole page instead of one +// per id. Each row carries its analysisId so callers can group results back per id; +// order within a group still follows d.downloads DESC NULLS LAST. +export async function getVerdictResultsBatch( + qx: QueryExecutor, + analysisIds: string[], +): Promise<(VerdictResultRow & { analysisId: string })[]> { + if (analysisIds.length === 0) { + return [] + } + const rows = await qx.select( + ` + SELECT + v.analysis_id, d.name, d.version, d.downloads, + v.reachable_verdict, v.confidence, v.evidence, v.reasoning + FROM blast_radius_verdicts v + JOIN blast_radius_dependents d ON d.id = v.dependent_id + WHERE v.analysis_id = ANY($(analysisIds)::uuid[]) + ORDER BY d.downloads DESC NULLS LAST + `, + { analysisIds }, + ) + return rows.map((row: Record) => ({ + ...(row as unknown as VerdictResultRow), + analysisId: row.analysis_id as string, + })) +} + // ---- stage runs (monitoring) ---- export async function startStageRun(qx: QueryExecutor, input: StageRunInput): Promise {