From 2464be1edf1cf1bf1ea96da7d6f2c3c6705427b1 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Thu, 23 Jul 2026 14:39:01 +0200 Subject: [PATCH 1/4] feat: implment bulk radius api Signed-off-by: Umberto Sgueglia --- .../public/v1/akrites-external/openapi.yaml | 63 +++++++++++++++++++ .../v1/packages/getBlastRadiusJobBatch.ts | 24 +++++++ .../v1/packages/submitBlastRadiusJobBatch.ts | 11 ++-- 3 files changed, 91 insertions(+), 7 deletions(-) diff --git a/backend/src/api/public/v1/akrites-external/openapi.yaml b/backend/src/api/public/v1/akrites-external/openapi.yaml index 034e53c766..84f195b75b 100644 --- a/backend/src/api/public/v1/akrites-external/openapi.yaml +++ b/backend/src/api/public/v1/akrites-external/openapi.yaml @@ -514,6 +514,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] diff --git a/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts b/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts index bdae178493..d88a6f37cd 100644 --- a/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts +++ b/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts @@ -69,3 +69,27 @@ export async function getBlastRadiusJobBatch(req: Request, res: Response): Promi ok(res, { page, pageSize, total, results }) } + +async function pollOneAnalysis( + qx: Awaited>, + requestedAnalysisId: string, +): Promise { + const analysis = await blastRadiusDal.getAnalysisDetail(qx, requestedAnalysisId) + if (!analysis) { + return { requestedAnalysisId, found: false, analysis: null } + } + + const done = analysis.status === 'done' + const [verdictRows, excludedByRangeCount] = done + ? await Promise.all([ + blastRadiusDal.getVerdictResults(qx, requestedAnalysisId), + blastRadiusDal.getDependentsExcludedByRangeCount(qx, requestedAnalysisId), + ]) + : [[], 0] + + return { + requestedAnalysisId, + found: true, + analysis: toBlastRadiusAnalysis(analysis, verdictRows, excludedByRangeCount), + } +} diff --git a/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts index 81ee8de51a..4a30cecd0f 100644 --- a/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts +++ b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts @@ -3,6 +3,7 @@ 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' @@ -31,9 +32,10 @@ export async function submitBlastRadiusJobBatch(req: Request, res: Response): Pr 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, body)), + jobs.map((body) => submitOneJob(qx, packagesTemporal, body)), ) res.status(202).json({ results }) @@ -41,6 +43,7 @@ export async function submitBlastRadiusJobBatch(req: Request, res: Response): Pr async function submitOneJob( qx: QueryExecutor, + packagesTemporal: Client, body: BlastRadiusJobRequest, ): Promise { const jobPackage = body.package ?? null @@ -61,12 +64,6 @@ async function submitOneJob( // must not reject the whole batch's Promise.all, only this job's entry. await blastRadiusDal.createAnalysis(qx, analysisInput) - // Acquired per job (inside the try), not once up front — getPackagesTemporalClient - // caches its connection in a module-level singleton, so this is cheap once - // connected, but a first-ever connection failure must fail this job's entry only, - // not reject the whole batch before any per-job try/catch is in play. - const packagesTemporal = await getPackagesTemporalClient() - await packagesTemporal.workflow.start('analyzeBlastRadius', { taskQueue: 'blast-radius-worker', workflowId: `${TemporalWorkflowId.BLAST_RADIUS_ANALYSIS}/${analysisId}`, From 7fb55abbdf776a6674568104d124721b8ff19104 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Thu, 23 Jul 2026 15:14:18 +0200 Subject: [PATCH 2/4] feat: add tests Signed-off-by: Umberto Sgueglia --- .../v1/packages/getBlastRadiusJobBatch.ts | 24 ------------------- 1 file changed, 24 deletions(-) diff --git a/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts b/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts index d88a6f37cd..bdae178493 100644 --- a/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts +++ b/backend/src/api/public/v1/packages/getBlastRadiusJobBatch.ts @@ -69,27 +69,3 @@ export async function getBlastRadiusJobBatch(req: Request, res: Response): Promi ok(res, { page, pageSize, total, results }) } - -async function pollOneAnalysis( - qx: Awaited>, - requestedAnalysisId: string, -): Promise { - const analysis = await blastRadiusDal.getAnalysisDetail(qx, requestedAnalysisId) - if (!analysis) { - return { requestedAnalysisId, found: false, analysis: null } - } - - const done = analysis.status === 'done' - const [verdictRows, excludedByRangeCount] = done - ? await Promise.all([ - blastRadiusDal.getVerdictResults(qx, requestedAnalysisId), - blastRadiusDal.getDependentsExcludedByRangeCount(qx, requestedAnalysisId), - ]) - : [[], 0] - - return { - requestedAnalysisId, - found: true, - analysis: toBlastRadiusAnalysis(analysis, verdictRows, excludedByRangeCount), - } -} From ec3601d66a3ac3ba9bfc39b32f33c7aee54c6658 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Thu, 23 Jul 2026 16:52:52 +0200 Subject: [PATCH 3/4] fix: add advisory cache Signed-off-by: Umberto Sgueglia --- .../public/v1/akrites-external/openapi.yaml | 24 +++--- .../src/api/public/v1/packages/blastRadius.ts | 57 ++++++++++++- .../v1/packages/getBlastRadiusJobBatch.ts | 11 +-- .../v1/packages/submitBlastRadiusJob.test.ts | 64 ++++++++++++++- .../v1/packages/submitBlastRadiusJob.ts | 25 +++++- .../submitBlastRadiusJobBatch.test.ts | 80 ++++++++++++++++++- .../v1/packages/submitBlastRadiusJobBatch.ts | 27 +++++-- .../src/packages/blastRadius.ts | 28 +++++++ 8 files changed, 284 insertions(+), 32 deletions(-) diff --git a/backend/src/api/public/v1/akrites-external/openapi.yaml b/backend/src/api/public/v1/akrites-external/openapi.yaml index 84f195b75b..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. @@ -1178,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..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 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..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 4a30cecd0f..6ebb461893 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', { @@ -98,12 +113,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..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. From 59c09c182a9eec55427015b56f19a4c2a7096285 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Mon, 3 Aug 2026 12:14:37 +0200 Subject: [PATCH 4/4] feat: update the cache Signed-off-by: Umberto Sgueglia --- .../src/api/public/v1/packages/blastRadius.ts | 18 ++++------ .../v1/packages/submitBlastRadiusJob.ts | 9 ++--- .../v1/packages/submitBlastRadiusJobBatch.ts | 35 ++++++++----------- .../src/packages/blastRadius.ts | 7 ++-- 4 files changed, 26 insertions(+), 43 deletions(-) diff --git a/backend/src/api/public/v1/packages/blastRadius.ts b/backend/src/api/public/v1/packages/blastRadius.ts index 72c535b661..68f8925feb 100644 --- a/backend/src/api/public/v1/packages/blastRadius.ts +++ b/backend/src/api/public/v1/packages/blastRadius.ts @@ -13,10 +13,8 @@ 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. +// 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 @@ -49,10 +47,8 @@ export interface BlastRadiusJobEntry { status: BlastRadiusJobStatus } -// 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. +// 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 @@ -69,10 +65,8 @@ export function toBlastRadiusJobEntry(params: { } } -// 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). +// 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: { diff --git a/backend/src/api/public/v1/packages/submitBlastRadiusJob.ts b/backend/src/api/public/v1/packages/submitBlastRadiusJob.ts index f60eebbdcb..3ff0b5e4a4 100644 --- a/backend/src/api/public/v1/packages/submitBlastRadiusJob.ts +++ b/backend/src/api/public/v1/packages/submitBlastRadiusJob.ts @@ -14,12 +14,9 @@ import { 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, 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. +// 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) diff --git a/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts index 6ebb461893..aa43cabed7 100644 --- a/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts +++ b/backend/src/api/public/v1/packages/submitBlastRadiusJobBatch.ts @@ -3,7 +3,6 @@ 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' @@ -19,27 +18,19 @@ import { 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 — 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. -// -// 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) const qx = await getPackagesQx() - const packagesTemporal = await getPackagesTemporalClient() const results: BlastRadiusJobEntry[] = await Promise.all( - jobs.map((body) => submitOneJob(qx, packagesTemporal, body)), + jobs.map((body) => submitOneJob(qx, body)), ) res.status(202).json({ results }) @@ -47,7 +38,6 @@ export async function submitBlastRadiusJobBatch(req: Request, res: Response): Pr async function submitOneJob( qx: QueryExecutor, - packagesTemporal: Client, body: BlastRadiusJobRequest, ): Promise { const jobPackage = body.package ?? null @@ -62,9 +52,8 @@ 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. + // 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, @@ -79,6 +68,12 @@ async function submitOneJob( // 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 + // caches its connection in a module-level singleton, so this is cheap once + // connected, but a first-ever connection failure must fail this job's entry only, + // not reject the whole batch before any per-job try/catch is in play. + const packagesTemporal = await getPackagesTemporalClient() + await packagesTemporal.workflow.start('analyzeBlastRadius', { taskQueue: 'blast-radius-worker', workflowId: `${TemporalWorkflowId.BLAST_RADIUS_ANALYSIS}/${analysisId}`, diff --git a/services/libs/data-access-layer/src/packages/blastRadius.ts b/services/libs/data-access-layer/src/packages/blastRadius.ts index fc3530d4c1..be4fedc9e4 100644 --- a/services/libs/data-access-layer/src/packages/blastRadius.ts +++ b/services/libs/data-access-layer/src/packages/blastRadius.ts @@ -177,11 +177,8 @@ 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. +// 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 },