diff --git a/backend/src/api/public/v1/akrites-external/openapi.yaml b/backend/src/api/public/v1/akrites-external/openapi.yaml index 034e53c766..998e5cfcf3 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. Auth0 now issues a dedicated Akrites Enclave M2M client with its own @@ -418,7 +420,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 @@ -445,8 +450,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. @@ -514,6 +521,69 @@ components: allOf: - $ref: '#/components/schemas/BlastRadiusAnalysis' + 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 enum: [high, medium, low] @@ -1115,10 +1185,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..68f8925feb 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,14 @@ 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 has to be for submit to reuse it instead of +// starting a new workflow — see getCachedJobEntry. force=true bypasses this. +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 +47,53 @@ 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', but a cache hit +// passes 'done' 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 — returns the +// cached job entry on a hit, or null on a cache miss or force=true. +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 bdae178493..e4e2d54afe 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' @@ -37,15 +38,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..3ff0b5e4a4 100644 --- a/backend/src/api/public/v1/packages/submitBlastRadiusJob.ts +++ b/backend/src/api/public/v1/packages/submitBlastRadiusJob.ts @@ -8,16 +8,33 @@ 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. +// 2a — submit a blast-radius analysis job. Always exactly one job per request, +// unless getCachedJobEntry returns a hit — then that's returned instead, with +// no new row and no workflow start. 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 +42,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 81ee8de51a..aa43cabed7 100644 --- a/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts +++ b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts @@ -12,21 +12,18 @@ import { validateOrThrow } from '@/utils/validation' import { type BlastRadiusJobEntry, type BlastRadiusJobRequest, + getCachedJobEntry, 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. +// array entry (each may hit the cache via getCachedJobEntry). 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 does not +// fail the whole batch — that job's entry comes back status: 'failed'. export async function submitBlastRadiusJobBatch(req: Request, res: Response): Promise { const { jobs } = validateOrThrow(blastRadiusJobBatchRequestSchema, req.body) @@ -55,10 +52,20 @@ async function submitOneJob( } try { + // Cache lookup is inside the try too, so a DB error here resolves this + // job's entry as 'failed' instead of rejecting the whole batch. + 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) // Acquired per job (inside the try), not once up front — getPackagesTemporalClient @@ -101,12 +108,12 @@ async function submitOneJob( // best-effort — the job's entry below still reports status: 'failed' } - 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..be4fedc9e4 100644 --- a/services/libs/data-access-layer/src/packages/blastRadius.ts +++ b/services/libs/data-access-layer/src/packages/blastRadius.ts @@ -177,6 +177,31 @@ export async function getAnalysisDetail( ) } +// Advisory-cache lookup for submit: most recent 'done' analysis within maxAgeDays. +// packageName uses IS NOT DISTINCT FROM since it's nullable and NULL = NULL is never true. +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.