diff --git a/backend/src/api/public/v1/akrites-external/openapi.yaml b/backend/src/api/public/v1/akrites-external/openapi.yaml index a1d1ef1540..e1b2ca4d7f 100644 --- a/backend/src/api/public/v1/akrites-external/openapi.yaml +++ b/backend/src/api/public/v1/akrites-external/openapi.yaml @@ -12,8 +12,10 @@ info: Packages, Advisories and Contacts endpoints are implemented. Blast Radius submit (2a) and poll (2b) are both implemented, backed by a 4-stage Temporal pipeline (intel, dependents, reachability, report) for npm - packages; other ecosystems fail fast with ECOSYSTEM_NOT_SUPPORTED. The - 7-day result cache is specced separately and not yet built. + packages; other ecosystems fail fast with ECOSYSTEM_NOT_SUPPORTED. Submit + reuses a 'done' analysis for the same advisory/package/ecosystem if it + completed within the last day (configurable, see the `force` field below) + instead of starting a new Temporal workflow. TODO: scopes below (read:packages, read:stewardships) are the existing @@ -57,10 +59,11 @@ tags: 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). + limiter as the single-job route, defaulting to 50 requests/hour + (configurable via AKRITES_BLAST_RADIUS_RATE_LIMIT_MAX / _WINDOW_MS); + 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: @@ -406,7 +409,10 @@ components: force: type: boolean default: false - description: Bypasses the 7-day cache and always triggers a new run. Use sparingly. + description: > + Bypasses the advisory cache (a 'done' analysis for the same + advisory/package/ecosystem completed within the last day, by + default) and always triggers a new run. Use sparingly. BlastRadiusJobEntry: type: object @@ -433,8 +439,10 @@ components: type: string enum: [pending, running, done, failed] description: > - Pending in the single-job submit response, always returned before the - Temporal workflow runs. In the batch submit response, a job whose + Pending when a job is freshly submitted, returned before the Temporal + workflow runs. Done when the advisory cache is reused instead — see + force above — in which case analysisId is the cached analysis's own + id, already completed. In the batch submit response, a job whose workflow failed to start comes back as failed instead — the rest of the batch is unaffected. @@ -1067,10 +1075,9 @@ paths: Starts a Temporal workflow running the 4-stage reachability pipeline (intel, dependents, reachability, report) for npm; other ecosystems fail fast with ECOSYSTEM_NOT_SUPPORTED. Poll status/results via - GET /jobs/{analysisId}. - - - Not yet implemented: the 7-day result cache and force-bypass semantics. + GET /jobs/{analysisId}. Reuses a 'done' analysis for the same + advisory/package/ecosystem completed within the last day (by + default) instead of starting a new workflow, unless force is true. tags: [Blast Radius] security: - M2MBearer: diff --git a/backend/src/api/public/v1/packages/blastRadius.ts b/backend/src/api/public/v1/packages/blastRadius.ts index 69c7c5b21a..72c535b661 100644 --- a/backend/src/api/public/v1/packages/blastRadius.ts +++ b/backend/src/api/public/v1/packages/blastRadius.ts @@ -1,5 +1,8 @@ import { z } from 'zod' +import * as blastRadiusDal from '@crowd/data-access-layer/src/packages/blastRadius' +import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' + // The reachability pipeline is npm-only for now — every other ecosystem (including // a missing one) is rejected by the schema below before the Temporal workflow is // triggered. @@ -10,6 +13,16 @@ export const SUPPORTED_BLAST_RADIUS_ECOSYSTEMS = ['npm'] as const // so it is NOT run through purlFieldSchema/normalizePurl like the other endpoints. const ADVISORY_ID_PATTERN = /^(GHSA-[0-9a-zA-Z]{4}-[0-9a-zA-Z]{4}-[0-9a-zA-Z]{4}|CVE-\d{4}-\d{4,})$/ +// How recent a 'done' analysis for the same (advisoryId, package, ecosystem) has to +// be for submit to reuse it instead of starting a new Temporal workflow — see +// getRecentDoneAnalysis. Configurable via env so it can be tuned without a redeploy; +// defaults to 1 day. force=true on the request always bypasses this cache. +const blastRadiusCacheMaxAgeDaysEnv = Number(process.env.AKRITES_BLAST_RADIUS_CACHE_MAX_AGE_DAYS) +export const BLAST_RADIUS_CACHE_MAX_AGE_DAYS = + Number.isSafeInteger(blastRadiusCacheMaxAgeDaysEnv) && blastRadiusCacheMaxAgeDaysEnv > 0 + ? blastRadiusCacheMaxAgeDaysEnv + : 1 + export const blastRadiusJobRequestSchema = z.object({ advisoryId: z .string() @@ -36,19 +49,57 @@ export interface BlastRadiusJobEntry { status: BlastRadiusJobStatus } -// Builds the 2a response body. The pipeline isn't implemented yet, so every freshly -// submitted job comes back pending — see analyzeBlastRadius in packages_worker. +// Builds the 2a response body. status defaults to 'pending' (a freshly submitted +// job — see analyzeBlastRadius in packages_worker) but a cache hit passes the +// cached analysis's own status (always 'done' — see getRecentDoneAnalysis) so the +// caller doesn't need to poll a job that's already finished. export function toBlastRadiusJobEntry(params: { analysisId: string advisoryId: string package: string | null ecosystem: BlastRadiusJobEcosystem + status?: BlastRadiusJobStatus }): BlastRadiusJobEntry { return { analysisId: params.analysisId, advisoryId: params.advisoryId, package: params.package, ecosystem: params.ecosystem, - status: 'pending', + status: params.status ?? 'pending', + } +} + +// Shared by submitBlastRadiusJob and submitBlastRadiusJobBatch — looks up a +// recent 'done' analysis for the same (advisoryId, package, ecosystem) and, if +// found, builds the job entry for it. Returns null on a cache miss or when +// force=true (which bypasses the cache entirely). +export async function getCachedJobEntry( + qx: QueryExecutor, + params: { + advisoryId: string + package: string | null + ecosystem: BlastRadiusJobEcosystem + force: boolean + }, +): Promise { + if (params.force) { + return null } + + const cached = await blastRadiusDal.getRecentDoneAnalysis( + qx, + { advisoryOsvId: params.advisoryId, packageName: params.package, ecosystem: params.ecosystem }, + BLAST_RADIUS_CACHE_MAX_AGE_DAYS, + ) + if (!cached) { + return null + } + + return toBlastRadiusJobEntry({ + analysisId: cached.id, + advisoryId: params.advisoryId, + package: params.package, + ecosystem: params.ecosystem, + status: 'done', + }) } diff --git a/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts b/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts index 2dd397be19..8f52e81b09 100644 --- a/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts +++ b/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts @@ -1,5 +1,6 @@ import type { Request, Response } from 'express' +import { groupBy } from '@crowd/common' import * as blastRadiusDal from '@crowd/data-access-layer/src/packages/blastRadius' import { getPackagesQx } from '@/db/packagesDb' @@ -34,15 +35,7 @@ export async function getBlastRadiusJobBatch(req: Request, res: Response): Promi 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 verdictsByAnalysisId = groupBy(verdictRows, (row) => row.analysisId) const excludedByRangeCountByAnalysisId = new Map( excludedByRangeCounts.map(({ analysisId, count }) => [analysisId, count]), ) diff --git a/backend/src/api/public/v1/packages/submitBlastRadiusJob.test.ts b/backend/src/api/public/v1/packages/submitBlastRadiusJob.test.ts index eb70ad1bd9..1fab59b719 100644 --- a/backend/src/api/public/v1/packages/submitBlastRadiusJob.test.ts +++ b/backend/src/api/public/v1/packages/submitBlastRadiusJob.test.ts @@ -3,10 +3,11 @@ import { describe, expect, it, vi } from 'vitest' import { submitBlastRadiusJob } from './submitBlastRadiusJob' -const { start, createAnalysis, failAnalysis } = vi.hoisted(() => ({ +const { start, createAnalysis, failAnalysis, getRecentDoneAnalysis } = vi.hoisted(() => ({ start: vi.fn().mockResolvedValue(undefined), createAnalysis: vi.fn().mockResolvedValue(undefined), failAnalysis: vi.fn().mockResolvedValue(undefined), + getRecentDoneAnalysis: vi.fn(), })) vi.mock('@/db/packagesTemporal', () => ({ @@ -20,12 +21,15 @@ vi.mock('@/db/packagesDb', () => ({ vi.mock('@crowd/data-access-layer/src/packages/blastRadius', () => ({ createAnalysis, failAnalysis, + getRecentDoneAnalysis, })) function mockReqRes(body: unknown) { start.mockClear() createAnalysis.mockClear() failAnalysis.mockClear() + getRecentDoneAnalysis.mockClear() + getRecentDoneAnalysis.mockResolvedValue(null) const req = { body } as unknown as Request @@ -152,4 +156,62 @@ describe('submitBlastRadiusJob', () => { }) expect(errorMessage).toBe('temporal unreachable') }) + + it('reuses a recent done analysis instead of starting a workflow', async () => { + const { req, res, start, status, json } = mockReqRes({ + advisoryId: 'GHSA-jf85-cpcp-j695', + ecosystem: 'npm', + }) + getRecentDoneAnalysis.mockResolvedValue({ + id: 'cached-analysis-id', + advisory_osv_id: 'GHSA-jf85-cpcp-j695', + package_name: null, + ecosystem: 'npm', + status: 'done', + error: null, + candidates_considered: 5, + started_at: '2026-07-01T00:00:00.000Z', + completed_at: '2026-07-01T01:00:00.000Z', + }) + + await submitBlastRadiusJob(req, res) + + expect(createAnalysis).not.toHaveBeenCalled() + expect(start).not.toHaveBeenCalled() + expect(status).toHaveBeenCalledWith(202) + expect(json).toHaveBeenCalledWith( + expect.objectContaining({ + analysisId: 'cached-analysis-id', + advisoryId: 'GHSA-jf85-cpcp-j695', + package: null, + ecosystem: 'npm', + status: 'done', + }), + ) + }) + + it('bypasses the cache and starts a new workflow when force is true, even with a recent done analysis', async () => { + const { req, res, start } = mockReqRes({ + advisoryId: 'GHSA-jf85-cpcp-j695', + ecosystem: 'npm', + force: true, + }) + getRecentDoneAnalysis.mockResolvedValue({ + id: 'cached-analysis-id', + advisory_osv_id: 'GHSA-jf85-cpcp-j695', + package_name: null, + ecosystem: 'npm', + status: 'done', + error: null, + candidates_considered: 5, + started_at: '2026-07-01T00:00:00.000Z', + completed_at: '2026-07-01T01:00:00.000Z', + }) + + await submitBlastRadiusJob(req, res) + + expect(getRecentDoneAnalysis).not.toHaveBeenCalled() + expect(createAnalysis).toHaveBeenCalledTimes(1) + expect(start).toHaveBeenCalledTimes(1) + }) }) diff --git a/backend/src/api/public/v1/packages/submitBlastRadiusJob.ts b/backend/src/api/public/v1/packages/submitBlastRadiusJob.ts index d92ace7b75..f60eebbdcb 100644 --- a/backend/src/api/public/v1/packages/submitBlastRadiusJob.ts +++ b/backend/src/api/public/v1/packages/submitBlastRadiusJob.ts @@ -8,16 +8,36 @@ import { getPackagesQx } from '@/db/packagesDb' import { getPackagesTemporalClient } from '@/db/packagesTemporal' import { validateOrThrow } from '@/utils/validation' -import { blastRadiusJobRequestSchema, toBlastRadiusJobEntry } from './blastRadius' +import { + blastRadiusJobRequestSchema, + getCachedJobEntry, + toBlastRadiusJobEntry, +} from './blastRadius' // 2a — submit a blast-radius analysis job. Always exactly one job per request. -// Every submission gets a fresh analysisId and status pending. +// Every submission gets a fresh analysisId and status pending, unless a 'done' +// analysis for the same (advisoryId, package, ecosystem) is still within the +// advisory cache window — see BLAST_RADIUS_CACHE_MAX_AGE_DAYS — in which case that +// cached analysis is returned instead, with no new row and no workflow start. +// force=true on the request skips this cache entirely. export async function submitBlastRadiusJob(req: Request, res: Response): Promise { const body = validateOrThrow(blastRadiusJobRequestSchema, req.body) const jobPackage = body.package ?? null const jobEcosystem = body.ecosystem + const qx = await getPackagesQx() + const cached = await getCachedJobEntry(qx, { + advisoryId: body.advisoryId, + package: jobPackage, + ecosystem: jobEcosystem, + force: body.force, + }) + if (cached) { + res.status(202).json(cached) + return + } + const analysisId = generateUUIDv4() // Create the pending row synchronously, before starting the workflow — otherwise a @@ -25,7 +45,6 @@ export async function submitBlastRadiusJob(req: Request, res: Response): Promise // blastRadiusStart's own createAnalysis call and get a 404 for a job that was, in // fact, accepted. blastRadiusStart's createAnalysis upserts the same row, so this // is safe to run again from the workflow. - const qx = await getPackagesQx() await blastRadiusDal.createAnalysis(qx, { id: analysisId, advisoryOsvId: body.advisoryId, diff --git a/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.test.ts b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.test.ts index e3f37ae73d..d0be4e37b2 100644 --- a/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.test.ts +++ b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.test.ts @@ -3,10 +3,11 @@ import { describe, expect, it, vi } from 'vitest' import { submitBlastRadiusJobBatch } from './submitBlastRadiusJobBatch' -const { start, createAnalysis, failAnalysis } = vi.hoisted(() => ({ +const { start, createAnalysis, failAnalysis, getRecentDoneAnalysis } = vi.hoisted(() => ({ start: vi.fn().mockResolvedValue(undefined), createAnalysis: vi.fn().mockResolvedValue(undefined), failAnalysis: vi.fn().mockResolvedValue(undefined), + getRecentDoneAnalysis: vi.fn(), })) vi.mock('@/db/packagesTemporal', () => ({ @@ -20,12 +21,15 @@ vi.mock('@/db/packagesDb', () => ({ vi.mock('@crowd/data-access-layer/src/packages/blastRadius', () => ({ createAnalysis, failAnalysis, + getRecentDoneAnalysis, })) function mockReqRes(body: unknown) { start.mockClear() createAnalysis.mockClear() failAnalysis.mockClear() + getRecentDoneAnalysis.mockClear() + getRecentDoneAnalysis.mockResolvedValue(null) const req = { body } as unknown as Request @@ -90,6 +94,24 @@ describe('submitBlastRadiusJobBatch', () => { expect(errorMessage).toBe('temporal unreachable') }) + it('still resolves the batch when failAnalysis itself throws after a workflow.start failure', 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')) + failAnalysis.mockRejectedValueOnce(new Error('db unreachable')) + + await expect(submitBlastRadiusJobBatch(req, res)).resolves.toBeUndefined() + + 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' }) + }) + it('rejects a batch containing an unsupported ecosystem without submitting any job', async () => { const { req, res, start } = mockReqRes({ jobs: [ @@ -121,4 +143,60 @@ describe('submitBlastRadiusJobBatch', () => { await expect(submitBlastRadiusJobBatch(req, res)).rejects.toThrow() expect(start).not.toHaveBeenCalled() }) + + it('reuses a recent done analysis for one job while starting a fresh workflow for the other', async () => { + const { req, res, start, json } = mockReqRes({ + jobs: [ + { advisoryId: 'GHSA-jf85-cpcp-j695', ecosystem: 'npm' }, + { advisoryId: 'GHSA-652q-gvq3-74qv', ecosystem: 'npm' }, + ], + }) + getRecentDoneAnalysis.mockResolvedValueOnce({ + id: 'cached-analysis-id', + advisory_osv_id: 'GHSA-jf85-cpcp-j695', + package_name: null, + ecosystem: 'npm', + status: 'done', + error: null, + candidates_considered: 5, + started_at: '2026-07-01T00:00:00.000Z', + completed_at: '2026-07-01T01:00:00.000Z', + }) + + await submitBlastRadiusJobBatch(req, res) + + expect(createAnalysis).toHaveBeenCalledTimes(1) + expect(start).toHaveBeenCalledTimes(1) + + const [{ results }] = json.mock.calls[0] + expect(results[0]).toMatchObject({ + analysisId: 'cached-analysis-id', + advisoryId: 'GHSA-jf85-cpcp-j695', + status: 'done', + }) + expect(results[1]).toMatchObject({ advisoryId: 'GHSA-652q-gvq3-74qv', status: 'pending' }) + }) + + it('bypasses the cache and starts a new workflow when force is true', async () => { + const { req, res, start } = mockReqRes({ + jobs: [{ advisoryId: 'GHSA-jf85-cpcp-j695', ecosystem: 'npm', force: true }], + }) + getRecentDoneAnalysis.mockResolvedValue({ + id: 'cached-analysis-id', + advisory_osv_id: 'GHSA-jf85-cpcp-j695', + package_name: null, + ecosystem: 'npm', + status: 'done', + error: null, + candidates_considered: 5, + started_at: '2026-07-01T00:00:00.000Z', + completed_at: '2026-07-01T01:00:00.000Z', + }) + + await submitBlastRadiusJobBatch(req, res) + + expect(getRecentDoneAnalysis).not.toHaveBeenCalled() + expect(createAnalysis).toHaveBeenCalledTimes(1) + expect(start).toHaveBeenCalledTimes(1) + }) }) diff --git a/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts index fa0d7b6f89..8c0d97c2ee 100644 --- a/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts +++ b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts @@ -13,6 +13,7 @@ import { validateOrThrow } from '@/utils/validation' import { type BlastRadiusJobEntry, type BlastRadiusJobRequest, + getCachedJobEntry, toBlastRadiusJobEntry, } from './blastRadius' import { blastRadiusJobBatchRequestSchema } from './blastRadiusBatch' @@ -20,7 +21,10 @@ 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), +// start — unless a 'done' analysis for the same (advisoryId, package, ecosystem) +// is still within the advisory cache window (see BLAST_RADIUS_CACHE_MAX_AGE_DAYS), +// in which case that entry reuses the cached analysis instead. 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. @@ -58,10 +62,21 @@ async function submitOneJob( } try { + // Cache lookup is inside the try too — like createAnalysis/workflow.start below, + // a DB error here must resolve this job's entry as 'failed', not reject the whole + // batch's Promise.all and 500 every other job in it. + const cached = await getCachedJobEntry(qx, { + advisoryId: body.advisoryId, + package: jobPackage, + ecosystem: jobEcosystem, + force: body.force, + }) + if (cached) { + return cached + } + // 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. + // same comment on submitBlastRadiusJob for why (avoids a poll-race 404). await blastRadiusDal.createAnalysis(qx, analysisInput) await packagesTemporal.workflow.start('analyzeBlastRadius', { @@ -87,16 +102,23 @@ async function submitOneJob( }) } 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. + // failing to start must not take the rest of the batch down with it. The + // failAnalysis call below is deliberately its own try/catch too — if marking + // the row failed also fails (e.g. DB unreachable), that must still resolve + // this job's entry rather than reject the whole Promise.all and 500 the batch. const errorMessage = err instanceof Error ? err.message : String(err) - await blastRadiusDal.failAnalysis(qx, analysisInput, errorMessage) + try { + await blastRadiusDal.failAnalysis(qx, analysisInput, errorMessage) + } catch { + // best-effort — the job is already being reported as failed below + } - return { + return toBlastRadiusJobEntry({ analysisId, advisoryId: body.advisoryId, package: jobPackage, ecosystem: jobEcosystem, status: 'failed', - } + }) } } diff --git a/services/libs/data-access-layer/src/packages/blastRadius.ts b/services/libs/data-access-layer/src/packages/blastRadius.ts index 030657f9ce..fc3530d4c1 100644 --- a/services/libs/data-access-layer/src/packages/blastRadius.ts +++ b/services/libs/data-access-layer/src/packages/blastRadius.ts @@ -177,6 +177,34 @@ export async function getAnalysisDetail( ) } +// Advisory-cache lookup for submit: finds the most recent 'done' analysis for the +// same (advisoryOsvId, packageName, ecosystem) triple completed within maxAgeDays, +// so a submit can reuse it instead of starting a new Temporal workflow. packageName +// uses IS NOT DISTINCT FROM since it's nullable (advisory-wide analyses have no +// package) and NULL = NULL is never true in plain SQL equality. +export async function getRecentDoneAnalysis( + qx: QueryExecutor, + input: { advisoryOsvId: string; packageName: string | null; ecosystem: string }, + maxAgeDays: number, +): Promise { + return qx.selectOneOrNone( + ` + SELECT + id, advisory_osv_id, package_name, ecosystem, status, error, + candidates_considered, started_at, completed_at + FROM blast_radius_analyses + WHERE advisory_osv_id = $(advisoryOsvId) + AND package_name IS NOT DISTINCT FROM $(packageName) + AND ecosystem = $(ecosystem) + AND status = 'done' + AND completed_at >= NOW() - make_interval(days => $(maxAgeDays)) + ORDER BY completed_at DESC + LIMIT 1 + `, + { ...input, maxAgeDays }, + ) +} + // 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.